GCP数据流是否支持python中的kafka IO？

Question

我正在尝试使用Kafka从Kafka主题中读取数据。python代码中的ReadFromKafka（）方法。我的代码如下所示：

from apache_beam.io.external import kafka
import apache_beam as beam

options = PipelineOptions()

with beam.Pipeline(options=options) as p:
           plants = (
      p
        |       'read' >> kafka.ReadFromKafka({'bootstrap.servers': 'public_ip:9092'}, ['topic1']))

但下面是错误消息。

错误：apache\u beam。跑步者。运行程序：访问读取回溯时出错（上次调用）：文件“test_File.py”，第16行，在

是因为apache光束数据流转轮不支持kafkaIO吗？




Answer 1

现在数据流支持多语言数据流Pipelinne检查此：链接

Answer 2

beam的python SDK不支持连接到Kafka。下面是一段代码片段

from __future__ import print_function
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from beam_nuggets.io import kafkaio

kafka_topic = "notifications"
kafka_config = {"topic": kafka_topic,
                "bootstrap_servers": "localhost:9092",
                "group_id": "notification_consumer_group"}

with beam.Pipeline(options=PipelineOptions()) as p:
    notifications = p | "Reading messages from Kafka" >> kafkaio.KafkaConsume(kafka_config)
    notifications | 'Writing to stdout' >> beam.Map(print)

bootstrap_服务器是一种逗号分隔的主机和端口配置，在这里部署代理。您将从Kafka群集配置中获得此信息。