带火花连接器的Cassandra-如何向Cassandra插入项目列表

Question

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

然后您只需编写as per文档：

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

附言。但是要小心，如果tar太大，在创建列表 时，可能会在workers上导致内存错误。根据tar文件中的文件格式，最好先解压缩数据，然后使用Spark读取数据。