如何使用Spark 2.2结构化流将Kafka流式数据写入Hive？

针对你现在的需求——用Spark 2.2 Structured Streaming处理Kafka每秒100条的批量数据，写入已建好的ORC格式Hive表demo_user，我整理了完整的实现方案和代码，帮你替换掉临时的静态插入逻辑。

先回顾下你的Hive表信息：

Hive表创建语句 #

CREATE TABLE demo_user(
 timeaa BIGINT,
 numberbb INT,
 decimalcc DOUBLE,
 stringdd STRING,
 booleanee BOOLEAN
) STORED AS ORC ;

手动插入语句 #

INSERT INTO TABLE demo_user VALUES (1514133139123, 14, 26.4, 'pravin', true);

完整的Spark Structured Streaming实现代码 #

下面是整合了Kafka数据读取、解析，以及流式写入Hive的完整代码：

import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions.*;
import org.apache.spark.sql.types.*;
import org.apache.spark.sql.streaming.Trigger;
import org.apache.spark.sql.SaveMode;

public class KafkaToHiveStreaming {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("KafkaToHiveDemo");
        conf.setMaster("local[2]"); // 生产环境请移除该本地模式配置
        conf.set("hive.metastore.uris", "thrift://localhost:9083");
        
        SparkSession session = SparkSession.builder()
                .config(conf)
                .enableHiveSupport()
                .getOrCreate();

        // 从Kafka读取数据并解析为与Hive表匹配的结构
        Dataset<Row> dataset = readFromKafka(session);

        // 流式写入Hive表的核心逻辑
        dataset.writeStream()
                .foreachBatch((batchDF, batchId) -> {
                    // 方案1：直接用DataFrame API写入Hive表
                    batchDF.write()
                            .mode(SaveMode.Append)
                            .format("orc")
                            .saveAsTable("demo_user");
                    
                    // 方案2：若需更灵活的SQL操作，可先创建临时视图再插入
                    // batchDF.createOrReplaceTempView("temp_demo_user");
                    // session.sql("INSERT INTO demo_user SELECT * FROM temp_demo_user");
                })
                .option("checkpointLocation", "/your/stable/checkpoint/path") // 必须指定，保证流式任务容错
                .trigger(Trigger.ProcessingTime("1 second")) // 每秒处理一批，匹配你的数据频率
                .start()
                .awaitTermination();
    }

    private static Dataset<Row> readFromKafka(SparkSession session) {
        return session.readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "your-kafka-broker:9092") // 替换为你的Kafka集群地址
                .option("subscribe", "xyz") // 订阅目标Kafka主题
                .load()
                // 将Kafka的value字段转为String（假设消息为JSON格式）
                .selectExpr("CAST(value AS STRING)")
                // 定义与Hive表结构完全匹配的Schema
                .select(from_json(col("value"), new StructType()
                        .add("timeaa", LongType)
                        .add("numberbb", IntegerType)
                        .add("decimalcc", DoubleType)
                        .add("stringdd", StringType)
                        .add("booleanee", BooleanType)
                ).as("parsed_data"))
                .select("parsed_data.*"); // 展开所有字段，得到与Hive表匹配的DataFrame
    }
}

关键注意事项 #

• Checkpoint路径必须指定：这是Structured Streaming实现容错和Exactly-Once语义的核心，生产环境要选择稳定的分布式存储路径（比如HDFS路径），不能用本地路径。
• Kafka消息解析适配：上面代码假设Kafka消息是JSON格式，如果你的消息是CSV或其他格式，需要调整解析逻辑。比如CSV格式可以用split拆分字段：

  .select(split(col("value"), ",").as("fields"))
  .select(
      col("fields").getItem(0).cast(LongType).as("timeaa"),
      col("fields").getItem(1).cast(IntegerType).as("numberbb"),
      col("fields").getItem(2).cast(DoubleType).as("decimalcc"),
      col("fields").getItem(3).as("stringdd"),
      col("fields").getItem(4).cast(BooleanType).as("booleanee")
  )
  

• 写入模式选择：SaveMode.Append保证每次微批数据追加到Hive表，不会覆盖已有数据，符合流式数据的写入需求。
• Trigger设置：Trigger.ProcessingTime("1 second")让Spark每秒处理一批数据，刚好匹配你Kafka主题每秒100条的批量数据频率。

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