与 Cassandra SaveToCassandra 一起火花

Question

我正在使用带有 spark 的 Cassandra，我有一个 JavaRDD<Srring> 包含 6 列，由 , 分隔，如下所示：

header: canal,action,time,tiemend,client

我创建了一个包含 6 列的表 Mytable：

CREATE TABLE IF NOT EXISTS ref_event_by_user_session (
    canal       TEXT,
    action      TEXT,
    time        timestamp,
    timeend     timestamp,
    Client      INT,
    PRIMARY KEY(canal, action, time, timeend)
);

现在我想使用 javaFunctions().saveToCassandra 将我的 JavaRDD 保存在我的 Cassandra 表中， 但我不知道如何使用它。你能告诉我怎么做吗？

Answer 1

我找到了解决方案。 如果它可以帮助某人:) 我只是用canal String, action String, time Date, timeend Date, Client String,创建了一个java类table2

然后：

JavaRDD<Table2> table2 = session_7.map(new Function<Tuple2<KeyTable2, ValueTable2>, Table2>() {
       @Override
       public Table2 call(Tuple2<KeyTable2, ValueTable2> keyTable2ValueTable2Tuple2) throws Exception {
           SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
           Date startdate = formatter.parse(keyTable2ValueTable2Tuple2._1.time.replace("+","."));
           Date endtdate = formatter.parse(keyTable2ValueTable2Tuple2._1.endtime.replace("+","."));
           Table2 t = new Table2(keyTable2ValueTable2Tuple2._2.canal,keyTable2ValueTable2Tuple2._2.motif,startdate,endtdate,keyTable2ValueTable2Tuple2._1.client);
           return t;
       }
   });

    javaFunctions(table2).writerBuilder("Schema", "Table", mapToRow(Table2.class)).saveToCassandra();

注意：即使不同变量是公共的，您也必须在类中添加 getter。