pyflink脚本，从Kafka读取json数据写入本地文件

下面是一个从 Kafka 读取 JSON 数据并将其写入本地文件的 PyFlink 脚本：

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.descriptors import Kafka, FileSystem, Json
from pyflink.table.udf import udf

# 定义函数用于将 JSON 字符串转换成 Python 字典
@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))
def json_to_map(s):
    import json
    return json.loads(s)

# 创建 StreamExecutionEnvironment 和 StreamTableEnvironment
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)
t_env = StreamTableEnvironment.create(env)

# 定义 Kafka 和 FileSystem 的连接信息
kafka_props = {
    "bootstrap.servers": "localhost:9092",
    "group.id": "test-group"
}

fs_props = {
    "fs.default-scheme": "file:///",
    "storage.path": "/path/to/output"
}

# 注册 Kafka 和 FileSystem 表
t_env \
    .connect( \
        Kafka()
        .version("universal")
        .topic("test-topic")
        .start_from_earliest()
        .properties(kafka_props)
    ) \
    .with_format( \
        Json()
        .json_schema("{type: 'object', properties: {id: {type: 'string'}, name: {type: 'string'}}}")
        .fail_on_missing_field(True)
        .ignore_parse_errors(True)
    ) \
    .with_schema( \
        DataTypes.ROW([DataTypes.FIELD("id", DataTypes.STRING()), DataTypes.FIELD("name", DataTypes.STRING())])
    ) \
    .register_table_source("kafka_source")

t_env \
    .connect( \
        FileSystem()
        .path("/path/to/output")
        .with_format(Json())
        .with_schema( \
            DataTypes.ROW([DataTypes.FIELD("id", DataTypes.STRING()), DataTypes.FIELD("name", DataTypes.STRING())])
        )
    ) \
    .create_temporary_table("fs_sink")

# 从 Kafka 读取数据，转换成 Python 字典后写入本地文件
t_env \
    .from_path("kafka_source") \
    .select(json_to_map("value").alias("data")) \
    .select("data['id'] as id, data['name'] as name") \
    .insert_into("fs_sink")

# 执行任务
t_env.execute("kafka_to_local_file")

在脚本中，我们首先定义了一个 UDF 函数 `json_to_map`，用于将 JSON 字符串转换成 Python 字典。接着，我们创建了 StreamExecutionEnvironment 和 StreamTableEnvironment，并设置了并行度为 1。

然后，我们定义了 Kafka 和 FileSystem 的连接信息，以及 JSON 格式的数据格式。我们使用 `register_table_source` 方法将 Kafka 表注册到 TableEnvironment 中，使用 `create_temporary_table` 方法创建一个临时的 FileSystem 表。

最后，我们从 Kafka 表中读取数据，并将其转换成 Python 字典后写入本地文件。我们使用 `insert_into` 方法将数据插入到 FileSystem 表中，使用 `execute` 方法执行任务。