实时计算 Flink版sql1.13 怎么把数据以json格式写入kafka？

在Flink SQL中，你可以使用INSERT INTO语句将数据写入Kafka。以下是一个示例：

INSERT INTO kafka_topic
SELECT * FROM source_table
FOR SYSTEM TIME AS OF '2022-01-01 00:00:00'
WITH ('topic.expr' = 'CONCAT(\"my_topic\", LPAD(cast(rowtime as string), 20, \"_\"))',
      'format.type' = 'json',
      'format.deriveTopic' = 'true');

在这个示例中，我们从source_table中选择所有的列，并将它们插入到Kafka的主题kafka_topic中。我们使用FOR SYSTEM TIME AS OF子句来指定数据的发送时间，这是Flink SQL的一个新功能，它可以让你在数据发送时指定准确的时间。

我们还使用了WITH子句来配置Kafka的参数。在这个例子中，我们使用了'topic.expr'和'format.type'两个参数。'topic.expr'参数用于动态生成Kafka的主题名，'format.type'参数用于指定数据的格式为JSON。

请注意，这只是一个基本的示例，你可能需要根据实际的需求来调整这个语句。例如，你可能需要添加更多的列，或者使用不同的数据类型。

要将数据以JSON格式写入Kafka，可以使用Flink SQL中的INSERT INTO语句。以下是一个示例：

INSERT INTO kafka_topic 
VALUES (...), (...), ...
FORMAT JSON;

其中，kafka_topic是目标Kafka主题的名称，(...)表示要插入的值。使用FORMAT JSON指定输出格式为JSON。