如何用Scala(Spark2.2)实现Spark SQL处理Kafka JSON事件并写入Hive

针对你用Scala 2.11结合Spark 2.2 Structured Streaming处理Kafka JSON事件、用Spark SQL操作后写入Hive的需求，我来逐个拆解你的问题：

1. JSON事件处理与Schema相关问题 #

从Kafka拿到的value是字符串格式，第一步就是把它解析成结构化的DataFrame。这里分情况说明：

手动指定Schema（生产环境必选） #

如果你的JSON事件Schema完全一致，强烈建议手动指定Schema——流处理场景下自动推断Schema有不少坑：比如需要先采样数据才能推断，后续Schema一旦变动会直接导致任务失败，而且性能也不如预定义Schema稳定。

手动指定Schema用StructType和StructField实现，举个实际例子：
假设你的JSON事件结构是{"user_id": 123, "event_type": "click", "event_time": 1620000000}，那Schema可以这么定义：

import org.apache.spark.sql.types._

// 定义JSON对应的Schema，nullable根据实际业务设置
val jsonSchema = StructType(Seq(
  StructField("user_id", IntegerType, nullable = false),
  StructField("event_type", StringType, nullable = false),
  StructField("event_time", LongType, nullable = false)
))

然后用from_json函数解析字符串列，展开成结构化字段：

import org.apache.spark.sql.functions._

// 基于你已有的Kafka连接代码，继续解析JSON
val parsedDF = df.selectExpr("CAST(value AS STRING)")
  .select(from_json(col("value"), jsonSchema).alias("event_data"))
  .select("event_data.*") // 把嵌套的结构体展开成顶层列

自动推断Schema（仅测试环境用） #

如果是测试或者快速验证场景，也可以让Spark自动推断Schema，但仅限非生产：

• 方法一：先读取一批静态的JSON样本数据，获取Schema后复用

// 从本地或HDFS的样本JSON文件推断Schema
val inferredSchema = spark.read.json("/tmp/sample-kafka-events.json").schema

// 用推断出的Schema解析流数据
val parsedDF = df.selectExpr("CAST(value AS STRING)")
  .select(from_json(col("value"), inferredSchema).alias("event_data"))
  .select("event_data.*")

• 方法二：直接让from_json自动推断（不推荐，流处理中可能因为初始无数据导致Schema为空）

val parsedDF = df.selectExpr("CAST(value AS STRING)")
  .select(from_json(col("value"), schema = null).alias("event_data"))
  .select("event_data.*")

2. 创建TempView后执行Spark SQL查询 #

解析出结构化DataFrame后，创建临时视图非常简单，之后就能像操作普通数据库表一样写SQL了：

第一步：创建临时视图 #

用createOrReplaceTempView（如果视图已存在就替换，比createTempView更实用）：

// 给解析后的DataFrame创建临时视图，命名为kafka_events
parsedDF.createOrReplaceTempView("kafka_events")

第二步：执行SQL查询 #

用spark.sql()方法执行SQL，返回的结果还是一个DataFrame，你可以继续处理或者直接写入Hive：

// 举个例子：统计不同事件类型的点击量，只保留event_time在最近1小时内的数据
val processedDF = spark.sql("""
  SELECT 
    event_type, 
    COUNT(user_id) as event_count,
    MAX(event_time) as latest_event_time
  FROM kafka_events
  WHERE event_time >= UNIX_TIMESTAMP() - 3600
  GROUP BY event_type
""")

额外补充：写入Hive的注意事项 #

Spark 2.2中Structured Streaming写入Hive需要注意几点：

1. 必须设置检查点目录，保证流任务的容错性
2. 选择合适的输出模式（Append/Complete/Update）：比如上面的聚合查询需要用Complete模式（因为要输出全量聚合结果）
3. 可以直接用table()方法指定Hive表名，Spark会自动关联Hive元数据

示例代码：

processedDF.writeStream
  .format("hive")
  .outputMode("Complete") // 聚合查询用Complete模式，普通过滤用Append
  .option("checkpointLocation", "/hdfs/path/to/checkpoint-dir") // 必须设置，路径要在分布式存储上
  .table("your_hive_db.target_events_table") // 写入指定Hive表
  .start()
  .awaitTermination()

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