使用用于 Scala 的 spark cassandra 连接器进行批量插入

Question

我正在尝试使用批处理语句将一些行插入 cassandra。我无法找到任何有关如何使用 datastax spark 连接器进行操作的文档。我已经尝试了以下

CassandraConnector(conf).withSessionDo({ session =>
    val ps = session.prepare(s"BEGIN BATCH" + 
    "INSERT INTO test.user_trans (user_id, amount) VALUES(?, ?);" +
    "INSERT INTO standing (user_id, position) VALUES (?, ?);" + 
    "APPLY BATCH" );
  val bound = ps.bind(user_id, amount, user_id, position)
  session.execute(bound)
});

但 datastax 驱动程序给出以下错误

com.datastax.driver.core.exceptions.SyntaxError: line 1:6 mismatched input 'BATCHINSERT' expecting K_BATCH (BEGIN [BATCHINSERT] INTO...)

是否可以使用 datastax spark 连接器驱动程序进行原子批量更新？

我使用的是 spark-cassandra-connector 版本 1.6.0-s_2.10 和火花 1.6.1

RDD 上的所有其他操作以及将单行保存到 Cassandra 都有效。

Answer 1

因为你忘了在BEGIN BATCH后面加一个空格

val ps = session.prepare(s"BEGIN BATCH " + // Put a space after BEGIN BATCH "INSERT INTO test.user_trans (user_id, amount) VALUES(?, ?);" + "INSERT INTO standing (user_id, position) VALUES (?, ?);" + "APPLY BATCH" );