基于 Apache Beam KafkaIO python SDK 处理 Avro 消息答案

【问题标题】：Handing Avro messages based on Apache Beam KafkaIO python SDK基于 Apache Beam KafkaIO python SDK 处理 Avro 消息
【发布时间】：2023-03-11 00:39:01
【问题描述】：

我目前正在尝试将消息作为 ByteDeserializer 读取，类似于以下示例 KafkaIO。我的测试设置如下：

选项 1：配置为使用 --runner=PortableRunner

选项 2：启动本地 flink 作业服务器，

docker run --net=host apache/beam_flink1.10_job_server:latest

发布测试 kafka avro 消息

管道参数定义为，

pipeline_args = ['--runner', 'FlinkRunner',
                 '--job_endpoint', 'localhost:8099',
                 '--environment_type', 'LOOPBACK',
                 '--flink_version', '1.10',
                 '--flink_master', 'localhost:8081']
pipeline_options = PipelineOptions(pipeline_args, save_main_session=True,streaming=True)

管道设置，

_ = (pipeline | ReadFromKafka(
                consumer_config= {'bootstrap.servers':'localhost:9092'},
                topic = ['beam-test-topic'])
              | beam.Flatmap(lambda kv: log_topic_contents(kv[1])))

我在执行 pipeline 时，使用默认的扩展服务 SDK 镜像（apache/beam_python3.7_sdk:2.29.0），并将作业提交到 flink 作业服务器。 flink 作业服务器失败并显示“无法提交 JobGraph”和“Rest 端点关闭”消息。

我是否会错过管道的任何运行时配置？

【问题讨论】：

标签： python apache-beam apache-beam-kafkaio

【解决方案1】：

“LOOPBACK”环境类型目前不支持跨语言转换。你能用 'DOCKER' 类型重试吗？

【讨论】：