Spark Structured Streaming能否用SQL向Hudi表插入数据？代码能否运行？

1. Spark Structured Streaming是否支持通过SQL向Hudi表执行INSERT操作？ #

在你使用的Hudi 0.12版本中，不支持直接通过SQL的INSERT INTO语句向流式读取的Hudi表执行写入。Hudi的流写入在该版本主要依赖writeStream API实现，而非SQL INSERT语法处理流数据写入逻辑。

2. 提供的Scala代码能否正常运行？ #

不能正常运行，报错核心原因如下：
你通过spark.readStream.format("hudi").load(...)创建的是流式读取的临时视图，而Spark SQL的INSERT INTO语句无法直接将流数据写入这类流式视图对应的Hudi表，这也是异常信息中unresolved operator 'InsertIntoStatement StreamingRelation'的含义——Spark无法解析针对流式关系的INSERT操作。

正确实现方案 #

要完成从Kafka流写入Hudi表的需求，需使用writeStream API替代SQL INSERT，示例代码如下：

// 先从Kafka流视图读取并处理数据
val processedDF = spark.sql(
  s"""
     |SELECT
     |UUID() AS `uuid`,
     |*,
     |DATE_FORMAT(FROM_UNIXTIME(`ts`/ 1000), 'yyyy-MM-dd') AS `dt`
     |FROM [KAFKA_TABLE_VIEW];
     |""".stripMargin
)

// 配置Hudi写入核心参数
val hudiOptions = Map(
  "hoodie.table.name" -> "ODS_LOG",
  "hoodie.datasource.write.recordkey.field" -> "uuid", // 替换为你的表主键字段
  "hoodie.datasource.write.partitionpath.field" -> "dt", // 替换为你的分区字段
  "hoodie.datasource.write.table.type" -> "COPY_ON_WRITE", // 可选MERGE_ON_READ
  "hoodie.datasource.write.operation" -> "upsert",
  "hoodie.datasource.write.precombine.field" -> "ts" // 替换为用于合并的时间字段
)

// 流式写入Hudi表
processedDF.writeStream
  .format("hudi")
  .options(hudiOptions)
  .option("checkpointLocation", "/path/to/your/checkpoint") // 必须指定检查点路径
  .start(getTablePath(ODS_LOG))
  .awaitTermination()

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