java - 从 Spark Java 将值附加到 Cassandra Map 列

Question

我有 3 列的 cassandra 表。

id text,
value text,
mappings map<text,text>

假设示例数据如下:

id        | value       | mappings
-----------------------------------------------
1ABC      | xyz         | {"a":"abc","b":"bcd"}

在 Spark 作业中，我将 id 1ABC 的 b 映射计算为 HashMap Ex: "b":"xyz"(可以将map转换为JavaRDD)

如何使用 cassandra java Spark 连接器将此值附加(覆盖)到表中？我在看this有关如何处理 CQL 集合附加的示例，但我似乎不知道如何在 Java 中执行此操作。

Answer 1

解决这个问题如下。

通过传递新参数或使用 Spark session 中的相同参数来创建 cassandra 连接器。

import com.datastax.spark.connector.cql.CassandraConnector;

CassandraConnector connector = CassandraConnector.apply(spark.sparkContext().conf()); // or pass different values for spark.cassandra.connection.host, username and password

rdd.foreach(new VoidFunction<TestBean>() {
        @Override
        public void call(TestBean t) throws Exception {
            final String id = t.getId();
            final Map<String, String> mappings = t.getMappings();
            boolean isUpdated = connector.withSessionDo(new AbstractFunction1<Session, Boolean>() {
                @Override
                public Boolean apply(Session v1) {
                    ResultSet updateResultSet = v1.execute(v1.prepare("update test set mappings = mappings + ? where id = ?")
                            .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM)
                            .bind(mappings, id));
                    return updateResultSet.wasApplied();
                }
            });
        }
    });