Flink流作业执行Merge Into语句报错，该语句是否受支持？

首先明确：当前Flink的Table SQL体系不支持针对Hudi的MERGE INTO语法，这就是你触发Unsupported query: Merge into异常的原因——Spark与Hudi的集成实现了该语法，但Flink侧的集成逻辑暂未支持。

要在Flink流作业中实现Hudi表的upsert操作，推荐以下两种方案：

方案1：使用Flink Table API + Hudi自动Upsert配置 #

Hudi本身支持根据主键自动判断插入/更新，只需在创建Hudi表时配置好主键和写模式，直接用INSERT INTO即可完成upsert。

步骤1：创建Hudi目标表（DDL） #

CREATE TABLE target (
  id STRING PRIMARY KEY NOT ENFORCED,
  amount DOUBLE
) WITH (
  'connector' = 'hudi',
  'path' = 'hdfs://your/hudi/target/path', -- 替换为实际存储路径
  'table.type' = 'MERGE_ON_READ', -- 可选COPY_ON_WRITE，根据业务场景选择
  'write.operation' = 'upsert', -- 开启自动upsert逻辑
  'hoodie.datasource.write.recordkey.field' = 'id' -- 指定主键字段，用于匹配更新
);

步骤2：写入数据 #

替换你原来的MERGE INTO代码，直接执行插入：

Table table = tableEnv.fromDataStream(KafkaStreamTableDataStreamStream);
tableEnv.createTemporaryView("table1", table);
// 直接插入，Hudi会自动处理upsert
tableEnv.executeSql("INSERT INTO target SELECT id, amount FROM table1");

方案2：使用Hudi Flink DataStream API #

如果需要更底层的控制逻辑，可直接使用DataStream API调用Hudi的写入接口：

// 构建Hudi写入配置
HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
    .withPath("hdfs://your/hudi/target/path")
    .withTableName("target")
    .withRecordKeyField("id")
    .withPreCombineField("ts") // 可选，用于多版本数据合并（比如取最新时间戳的记录）
    .build();

// 初始化Flink写入客户端
FlinkWriteClient<HoodieRecord> writeClient = new FlinkWriteClient<>(tableEnv.getConfig(), writeConfig);

// 将Kafka流转换为HudiRecord流
DataStream<HoodieRecord> hoodieStream = KafkaStreamTableDataStreamStream.map(row -> {
    String id = row.getFieldAs("id");
    Double amount = row.getFieldAs("amount");
    Map<String, Object> data = new HashMap<>();
    data.put("id", id);
    data.put("amount", amount);
    return HoodieRecord.newBuilder()
        .setRecordKey(new HoodieRecordKey(id))
        .setPayload(new OverwriteWithLatestAvroPayload(new GenericRecordAvroPayload(data)))
        .build();
});

// 执行写入
writeClient.startCommit();
hoodieStream.addSink(new FlinkHoodieSink<>(writeConfig, writeClient));

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