PyFlink - JSON 文件接收器？答案

【问题标题】：PyFlink - JSON file sink?PyFlink - JSON 文件接收器？
【发布时间】：2020-11-07 14:19:48
【问题描述】：

是否可以像使用 CSV 一样在 Table API 和/或 DataStream API 中使用 JSON 文件接收器？

谢谢！

代码

my_sink_ddl = f"""
    create table mySink (
        id STRING,
        dummy_item STRING
    ) with (
        'connector.type' = 'filesystem',
        'format.type' = 'json',
        'connector.path' = 'output.json'
    )
"""

错误

TableException: findAndCreateTableSink failed.

【问题讨论】：

标签： apache-flink pyflink

【解决方案1】：

是的，根据 Jira FLINK-17286 Integrate json to file system connector 和对应的 pull request [FLINK-17286][connectors / filesystem]Integrate json to file system connector #12010，可以从 Flink 1.11 开始。 Flink之前1.11我相信是不支持的。

您需要使用以下配置：

... with (
        'connector' = 'filesystem',
        'format' = 'json',
        'path' = 'output_json' -- This must be a directory
    )

加上以下环境定义：

t_env = BatchTableEnvironment.create(   environment_settings=EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build())

【讨论】：

感谢您的回复米卡莱。知道如何正确指定接收器吗？
请看一下这个 1.11 flink 的文档，它可能会有所帮助：ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/… 从我现在的情况来看 - 它应该与 ('connector' = 'filesystem', 'format' = ' json', 'path' = 'output.json')
现在使用这些属性名称会导致另一个错误：TableException: BatchTableSink or OutputFormatTableSink required to emit batch Table。
请确认您使用的是 flink 1.11 吗？
你能试试这个吗？ t_env = BatchTableEnvironment.create( environment_settings=EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build())