flink sql 按照指定字段分区写入kafka

可以通过 Flink SQL 中的 `PARTITION BY` 子句来指定分区字段，然后使用 Flink 的 Kafka Producer 将数据发送到 Kafka 中。下面是一个示例代码：

// 创建 Flink Table Environment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

// 创建 Kafka Producer 配置
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");

// 定义输入数据源
String sourceDDL = "CREATE TABLE source_table (id INT, name STRING, event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND) WITH (...)";
tEnv.executeSql(sourceDDL);

// 定义输出数据源
String sinkDDL = "CREATE TABLE sink_table (id INT, name STRING) WITH ('connector' = 'kafka', 'topic' = 'output-topic', 'properties' = '" + kafkaProps.toString() + "', 'format' = 'json')";
tEnv.executeSql(sinkDDL);

// 执行 SQL 查询并写入 Kafka
String sql = "INSERT INTO sink_table SELECT id, name FROM source_table PARTITION BY id";
tEnv.executeSql(sql);

在上面的代码中，我们首先创建了一个 Flink Table Environment，并且定义了 Kafka Producer 的配置。然后，我们使用 Flink SQL 创建了输入和输出表。输入表包括一个 `event_time` 字段，我们使用它来定义 watermark。输出表是一个 Kafka topic，我们使用 `PARTITION BY` 子句按照 `id` 字段进行分区。最后，我们执行了 SQL 查询并将结果写入 Kafka topic。