Spark中如何将Dataset<Row>转为键值对双列结构输出至Kafka？

问题原因 #

你最后一个map操作使用了Encoders.javaSerialization(Row.class)，这种序列化方式会把整个Row对象当成单一的value字段处理，Spark无法识别Row内部的两个元素结构，因此最终的DataFrame只有一列。

修正方案（推荐简洁版） #

直接将Tuple2<String, String>类型的Dataset转换为带有key和value列的DataFrame，无需额外的map操作：

Dataset<Row> df = spark.readStream().format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", "cab-location")
        .option("startingOffsets", "earliest").load();

// 处理数据并直接转为带key、value列的DataFrame
Dataset<Row> outputDF = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
        .map(new MapFunction<Row, Tuple2<String, String>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, String> call(Row value) throws Exception {
                Gson g = new Gson();  
                CabLocationData cabLocationData = g.fromJson(value.getString(1), CabLocationData.class);
                return new Tuple2<>(value.getString(0), cabLocationData.getCabName());
            }
        }, Encoders.tuple(Encoders.STRING(), Encoders.STRING()))
        .toDF("key", "value"); // 指定列名，生成包含key和value的DataFrame

// 验证列名
System.out.println(Arrays.toString(outputDF.columns())); // 输出 [key, value]

写入Kafka主题 #

生成符合要求的DataFrame后，即可直接写入目标Kafka主题：

outputDF.writeStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("topic", "your-output-topic") // 替换为你的输出主题名
        .option("checkpointLocation", "/path/to/checkpoint") // 流式处理必须指定检查点路径
        .start()
        .awaitTermination();

备选方案（显式定义Row Schema） #

如果必须通过Row转换，可通过定义StructType来明确列结构，避免使用javaSerialization：

// 定义输出DataFrame的Schema
StructType outputSchema = new StructType()
        .add("key", DataTypes.StringType)
        .add("value", DataTypes.StringType);

Dataset<Row> df = spark.readStream().format("kafka")
        .option("kafka.bootstrap.servers", "kafka:9092")
        .option("subscribe", "cab-location")
        .option("startingOffsets", "earliest").load();

Dataset<Row> outputDF = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
        .map(new MapFunction<Row, Tuple2<String, String>>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Tuple2<String, String> call(Row value) throws Exception {
                Gson g = new Gson();  
                CabLocationData cabLocationData = g.fromJson(value.getString(1), CabLocationData.class);
                return new Tuple2<>(value.getString(0), cabLocationData.getCabName());
            }
        }, Encoders.tuple(Encoders.STRING(), Encoders.STRING()))
        .map(new MapFunction<Tuple2<String, String>, Row>() {
            private static final long serialVersionUID = 1L;

            @Override
            public Row call(Tuple2<String, String> value) throws Exception {
                return RowFactory.create(value._1, value._2);
            }
        }, Encoders.bean(Row.class, outputSchema)); // 使用指定Schema的Encoder

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