Cassandra 与 Spark 连接器 - 如何将项目列表插入 Cassandra

Question

使用 Java 使用 Cassandra 和 Spark 2.12 (3.2.0)。 Cassandra 连接器 3.1.0

我从 s3 获取的目的是进行预处理并将并行插入 Cassandra。

我遇到的问题是我对每个 s3 文件进行了预处理，其中包括要插入 Cassandra 的项目列表，如下所示：JavaRDD<List<SearchEntity>>

我应该如何将它传递给 cassandra（如代码示例中所示）？我看到它支持单个对象mapToRow。 也许我错过了什么？

使用以下代码

public static void main(String[] args) throws Exception {
    SparkConf conf = new SparkConf()
        .setAppName("Example Spark App")
        .setMaster("local[1]")
        .set("spark.cassandra.connection.host", "127.0.0.1");
        
    JavaSparkContext sparkContext = new JavaSparkContext(conf);
    sparkContext.hadoopConfiguration().set("fs.s3a.access.key", "XXXX");
    sparkContext.hadoopConfiguration().set("fs.s3a.secret.key", "YYYY");
    sparkContext.hadoopConfiguration().set("fs.s3a.endpoint", "XXXXX");
    sparkContext.hadoopConfiguration().set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem");
    sparkContext.hadoopConfiguration().set("mapreduce.input.fileinputformat.input.dir.recursive", "true");
                
    JavaPairRDD<String, PortableDataStream> javaPairRDD = sparkContext.binaryFiles("s3a://root/folder/");
        
    File ROOT = createTempFolder().getAbsoluteFile();
        
    JavaRDD<List<SearchEntity>> listJavaRDD = javaPairRDD.map(rdd -> {
            System.out.println("Working on TAR: " + rdd._1);
        
            DataInputStream stream = rdd._2.open();
        
            // some preprocess
            List<SearchEntity> toCassandraList = new WorkerTest(ROOT, stream).run();
        
            return toCassandraList;
        });
        
    // here I want to take List<SearchEntity> toCassandraList and save them
    // but I don't see how as it support only single object ..
    CassandraJavaUtil.javaFunctions(listJavaRDD)
        .writerBuilder("demoV2", "simple_search", 
                       CassandraJavaUtil.mapToRow(List<SearchEntity> list objects ...)) // here is problem
        .saveToCassandra();
        
    System.out.println("Finish run s3ToCassandra:");
    sparkContext.stop();
}

架构之前是手动配置的，仅用于测试目的。

CREATE TABLE simple_search (
    engine text,
    term text,
    time bigint,
    rank bigint,
    url text,
    domain text,
    pagenum bigint,
    descr text,
    display_url text,
    title text,
    type text,
    PRIMARY KEY ((engine, term), time , url, domain, pagenum)
) WITH CLUSTERING ORDER BY 
  (time DESC, url DESC,  domain DESC , pagenum DESC);

欢迎使用 Java 和 Scala 解决方案

Answer 1

要写入数据，您需要在SearchEntity 上工作，而不是在SearchEntity 的列表上工作。为此，您需要使用flatMap 而不是普通的map：

JavaRDD<SearchEntity> entriesRDD = javaPairRDD.flatMap(rdd -> {
        System.out.println("Working on TAR: " + rdd._1);
        DataInputStream stream = rdd._2.open();
        // some preprocess
        List<SearchEntity> toCassandraList = new WorkerTest(ROOT, stream).run();
        return toCassandraList;
    });

然后你就可以按照documentation：

javaFunctions(rdd).writerBuilder("demoV2", "simple_search",
   mapToRow(SearchEntity.class)).saveToCassandra();

附：但请注意，如果您的 tar 太大，在创建 List<SearchEntity> 时可能会导致工作人员出现内存错误。根据 tar 文件中的文件格式，最好先解压缩数据，然后使用 Spark 读取它们。