如何基于单个Kafka Topic事件用Java-Spark写入多Hive表？

核心思路 #

利用Spark的DataFrame过滤能力，将单Topic消费到的消息按事件类型（如原Oracle表标识）拆分为多个数据流，分别写入对应Hive表。和多Topic场景的区别在于，多Topic是从源头拆分数据流，单Topic则是消费后通过过滤逻辑完成拆分。

实现步骤&代码示例 #

1. 初始化SparkSession（启用Hive支持）

import org.apache.spark.sql.SparkSession;

SparkSession spark = SparkSession.builder()
    .appName("KafkaSingleTopicToHive")
    .enableHiveSupport() // 必须启用该配置才能操作Hive表
    .getOrCreate();

2. 消费Kafka单Topic消息

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

Dataset<Row> kafkaRawDF = spark.readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka-broker-1:9092,kafka-broker-2:9092")
    .option("subscribe", "your-oracle-sync-topic") // 指定目标单Topic
    .load();

3. 解析消息并提取事件标识

假设Kafka消息为JSON格式，生产者已在消息中添加source_table字段标记原Oracle表（如table1、table2），业务数据存于payload字段：

import org.apache.spark.sql.functions.*;
import org.apache.spark.sql.types.IntegerType;
import org.apache.spark.sql.types.LongType;
import org.apache.spark.sql.types.DoubleType;

// 将Kafka的value字段转为字符串格式
Dataset<Row> jsonDF = kafkaRawDF.selectExpr("CAST(value AS STRING) AS json_content");

// 解析JSON，拆分出源表标识和业务数据
Dataset<Row> parsedDF = jsonDF.select(
    get_json_object(col("json_content"), "$.source_table").alias("source_table"),
    get_json_object(col("json_content"), "$.payload").alias("payload")
);

4. 按事件类型拆分数据流并写入Hive表

// 处理table1的数据，解析payload为Hive表对应字段结构
Dataset<Row> table1DF = parsedDF.filter(col("source_table").equalTo("table1"))
    .select(
        get_json_object(col("payload"), "$.id").cast(IntegerType).alias("id"),
        get_json_object(col("payload"), "$.user_name").alias("user_name"),
        get_json_object(col("payload"), "$.create_time").alias("create_time")
    );

// 写入hive.table1（流式写入需指定独立checkpoint目录）
table1DF.writeStream()
    .format("hive")
    .option("checkpointLocation", "/hdfs/path/checkpoint/table1")
    .mode("append")
    .table("hive.table1");

// 处理table2的数据，同理解析对应字段
Dataset<Row> table2DF = parsedDF.filter(col("source_table").equalTo("table2"))
    .select(
        get_json_object(col("payload"), "$.order_id").cast(LongType).alias("order_id"),
        get_json_object(col("payload"), "$.goods_id").cast(IntegerType).alias("goods_id"),
        get_json_object(col("payload"), "$.amount").cast(DoubleType).alias("amount")
    );

// 写入hive.table2
table2DF.writeStream()
    .format("hive")
    .option("checkpointLocation", "/hdfs/path/checkpoint/table2")
    .mode("append")
    .table("hive.table2");

// 启动流任务并等待结束
spark.streams().awaitAnyTermination();

关键注意事项 #

• 事件标识可靠性：必须确保生产者在消息中携带准确的表标识字段（如source_table），这是拆分逻辑的核心依据。
• 字段类型匹配：解析业务数据时，要严格对应Hive表的字段类型，避免因类型不兼容导致写入失败。
• Checkpoint隔离：每个表的流写入必须使用独立的checkpoint目录，防止不同任务的元数据冲突。
• 批量场景适配：如果是批量消费（而非流式），将writeStream替换为write，并指定mode("append")或mode("overwrite")即可。

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