使用批处理语句插入 Cassandra

Question

我正在使用批处理语句将来自 csv 文件的数据插入 Cassandra。我的桌子看起来像这样 创建表曝光(expoid bigint,fileid bigint,studyid text, projname text, w text, x text, y text, z text)

System.out.println(colDataMap);

String  keyspace = "orchtablespaces";
String tabName = fileName;
//String tableFile = "/home/blr-lt-202/empAccount.txt";
String tableFile = fname;

Set<String> colNamesSet = colDataMap.keySet();
String[] colNames = colNamesSet.toArray(new String[colNamesSet.size()]);
System.out.println("ColNames ::" +colNames);

String makeStatement =  makeSt(keyspace,tabName,colNames);

System.out.println("makeStatement ::"+makeStatement);

if(count==65534)
{
    session.executeAsync(bs);
    count = 0;
    bs = new BatchStatement();
}

PreparedStatement statement = session.prepare(makeStatement);
//bcz expo id and seq_n
String expoid =(String) colDataMap.get("expoid");
String Seq_No = (String) colDataMap.get("Seq_No");
colDataMap.put(expoid, Long.valueOf(expoid));
colDataMap.put("Seq_No", Long.valueOf(Seq_No));

BoundStatement query = statement.bind(colDataMap.values().toArray(new Object[colDataMap.size()]));

//BoundStatement query = statement.bind(colDataMap.get("Seq_No"),colDataMap.get("fileId"),colDataMap.get("studyId"),colDataMap.get("projectName"),colDataMap.get("dosetxt"),colDataMap.get("sdurtunit"),colDataMap.get("durtunit"));
System.out.println("query "+query);
bs.add(query);
count++;
}
session.executeAsync(bs);

session.close();

MapColDataMap 将所有 Cassandra 表列名称作为键及其值在如下所示的映射中

{Seq_No=0, fileId=123, studyId=786, w=PreCLinic, x=Tasq 30mg/kg, y=12 days, z=12 days}

Set<String> colNamesSet =colDataMap.keySet();
String[] colNames = colNamesSet.toArray(new String[colNamesSet.size()]);
System.out.println("ColNames ::" +colNames);

String makeStatement =  makeSt(keyspace,tabName,colNames);

System.out.println("makeStatement ::"+makeStatement);

方法makeSt准备准备好的Statement

makeStatement ::INSERT INTO orchtablespaces.EXPOSURES ( Seq_No,fileId,studyId,w,x,y,z ) values ( ?,?,?,?,?,?,? )

这一切都很好。然后我执行以下操作

PreparedStatement statement = session.prepare(makeStatement);
//bcz expoid and seq_no are of type bigint int the cassandra table
String expoid =(String) colDataMap.get("expoid");
String Seq_No = (String) colDataMap.get("Seq_No");
colDataMap.put(expoid, Long.valueOf(expoid));
colDataMap.put("Seq_No", Long.valueOf(Seq_No));

BoundStatement query = statement.bind(colDataMap.values().toArray(new Object[colDataMap.size()]));

当我运行程序时出现以下错误

com.datastax.driver.core.exceptions.NoHostAvailableException.copy(NoHostAvailableException.java:84)
com.datastax.driver.core.DefaultResultSetFuture.extractCauseFromExecutionException(DefaultResultSetFuture.java:289)com.test.load.microarr.CopyOfLoadMicroArr.transformSourceFile(CopyOfLoadMicroArr.java:486)com.test.load.microarr.CopyOfLoadMicroArr.loadData(CopyOfLoadMicroArr.java:149)
com.test.load.microarr.CopyOfLoadMicroArr.main(CopyOfLoadMicroArr.java:114)
Caused by: com.datastax.driver.core.exceptions.NoHostAvailableException: All host(s) tried for query failed (no host was tried)
com.datastax.driver.core.RequestHandler.sendRequest(RequestHandler.java:107)
com.datastax.driver.core.SessionManager.execute(SessionManager.java:538)
com.datastax.driver.core.SessionManager.prepareAsync(SessionManager.java:124)
com.datastax.driver.core.AbstractSession.prepare(AbstractSession.java:77)
    ... 3 more
com.datastax.driver.core.AbstractSession.prepare(AbstractSession.java:79)

程序中断的确切行是第 486 行，即在

 PreparedStatement statement = session.prepare(makeStatement);

我正在使用 DataStax Cassandra 2.1.8 .. 任何建议和解决方案都会非常有帮助！谢谢！

Answer 1

你肯定对与 cassandra 的连接有问题。

形成Datastax Java driver documentation：

NoHostAvailableException - 如果无法成功联系集群中的任何主机来准备此语句。

并形成the error description

由于没有主机而无法执行查询时引发的异常 可用的。如果

• 要么没有主机住 查询时的集群

• 所有已经尝试过的主机 由于连接问题而失败

出于调试目的， 已尝试的主机列表以及失败原因可以是 使用errors方法检索。

尝试捕获此错误并输出已尝试主机的列表 - 这可以帮助您检测是否存在错误的主机或错误的端口。使用getErrors()方法——http://docs.datastax.com/en/drivers/java/2.0/com/datastax/driver/core/exceptions/NoHostAvailableException.html#getErrors()