Добавление значений в столбец «Карта Кассандры» из Spark Java - PullRequest
Новая
       25

Добавление значений в столбец «Карта Кассандры» из Spark Java

0 голосов
/ 30 января 2020

У меня есть таблица cassandra с 3 столбцами. 

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

Допустим, пример данных выглядит следующим образом: 

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

В задании на искривление я вычислил новое значение для id * b 1ABC, отображаемое как HashMap Ex: "b":"xyz" (может преобразовать карту в JavaRDD)

Как добавить (перезаписать) это значение в таблицу, используя искровой соединитель cassandra java? Я смотрел на этот пример того, как обрабатывать добавление коллекций CQL, но я не могу понять, как это сделать в Java. Любые указатели будут оценены.

1 Ответ

1 голос
/ 05 февраля 2020

Решено, как показано ниже.

Создайте соединитель кассандры, передавая новые параметры или используя его из сеанса Spark. 

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();
                }
            });
        }
    });
...