Spark Cassandra连接器如何将JavaRDD<List<SearchEntity>>写入Cassandra

核心思路 #

• 当前RDD元素为List<SearchEntity>，而Cassandra连接器要求RDD元素为单条表记录对应的实体/行对象，只需通过flatMap操作把嵌套列表打平为JavaRDD<SearchEntity>即可适配写入API。
• 只要SearchEntity是标准POJO且字段名、类型与Cassandra表结构完全匹配，可直接使用连接器自带的POJO映射能力，无需手动编写行转换逻辑。

前置要求 #

SearchEntity类实现Serializable接口，符合Java Bean规范：包含无参构造函数，所有字段有对应的getter/setter方法，字段名与Cassandra表列名完全一致。

修改后的完整代码 #

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")
        // 可选：优化Cassandra写入性能
        .set("spark.cassandra.output.batch.size.rows", "2000")
        .set("spark.cassandra.output.concurrent.writes", "10");
        
    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;
        });
    
    // 新增：把嵌套列表打平为单实体RDD
    JavaRDD<SearchEntity> entityRDD = listJavaRDD.flatMap(list -> list.iterator());
        
    // 直接使用POJO映射写入Cassandra
    CassandraJavaUtil.javaFunctions(entityRDD)
        .writerBuilder("demoV2", "simple_search", CassandraJavaUtil.mapToRow(SearchEntity.class))
        .saveToCassandra();
        
    System.out.println("Finish run s3ToCassandra:");
    sparkContext.stop();
}

特殊场景适配 #

如果SearchEntity字段名和Cassandra表列名不匹配，可自定义RowMapper手动映射字段：

RowMapper<SearchEntity> customMapper = entity -> RowFactory.create(
    entity.getEngine(),
    entity.getTerm(),
    entity.getTime(),
    entity.getRank(),
    entity.getUrl(),
    entity.getDomain(),
    entity.getPagenum(),
    entity.getDescr(),
    entity.getDisplayUrl(),
    entity.getTitle(),
    entity.getType()
);

// 写入时传入自定义Mapper
CassandraJavaUtil.javaFunctions(entityRDD)
    .writerBuilder("demoV2", "simple_search", customMapper)
    .saveToCassandra();

内容的提问来源于stack exchange，提问作者VitalyT