gcp数据流支持python中的kafka io吗？

rm5edbpk 于 2021-06-04 发布在 Kafka

关注(0)|答案(2)|浏览(515)

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

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']))

但下面是错误信息。 ERROR:apache_beam.runners.runner:Error while visiting read Traceback (most recent call last): File "test_file.py", line 16, in <module> | 'read' >> kafka.ReadFromKafka({'bootstrap.servers': 'localhost:9092'}, ['topic1']) File "/usr/local/lib/python3.7/dist-packages/apache_beam/pipeline.py", line 547, in __exit__ self.run().wait_until_finish() File "/usr/local/lib/python3.7/dist-packages/apache_beam/pipeline.py", line 526, in run return self.runner.run_pipeline(self, self._options) File "/usr/local/lib/python3.7/dist-packages/apache_beam/runners/dataflow/dataflow_runner.py", line 565, in run_pipeline self.visit_transforms(pipeline, options) File "/usr/local/lib/python3.7/dist-packages/apache_beam/runners/runner.py", line 224, in visit_transforms pipeline.visit(RunVisitor(self)) File "/usr/local/lib/python3.7/dist-packages/apache_beam/pipeline.py", line 572, in visit self._root_transform().visit(visitor, self, visited) File "/usr/local/lib/python3.7/dist-packages/apache_beam/pipeline.py", line 1075, in visit part.visit(visitor, pipeline, visited) File "/usr/local/lib/python3.7/dist-packages/apache_beam/pipeline.py", line 1078, in visit visitor.visit_transform(self) File "/usr/local/lib/python3.7/dist-packages/apache_beam/runners/runner.py", line 219, in visit_transform self.runner.run_transform(transform_node, options) File "/usr/local/lib/python3.7/dist-packages/apache_beam/runners/runner.py", line 249, in run_transform (transform_node.transform, self)) NotImplementedError: Execution of [<ReadFromKafka(PTransform) label=[ReadFromKafka(beam:external:java:kafka:read:v1)]>] not implemented in runner <apache_beam.runners.dataflow.dataflow_runner.DataflowRunner object at 0x7f72463344a8>. 是因为apache beam dataflow runner不支持kafkaio吗？

apache-kafka apache-beam google-cloud-dataflow apache-beam-io

来源：https://stackoverflow.com/questions/62775435/does-gcp-dataflow-support-kafka-io-in-python

2条答案

按热度按时间

mlmc2os51#

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\u服务器是一个逗号分隔的主机和端口配置，在这里部署代理。您将从kafka集群配置中获得此信息。

赞(0）回复(0）举报 2021-06-05

kx1ctssn2#

下面是一个好消息，现在dataflow支持多语言dataflow pipeline检查这个：link

赞(0）回复(0）举报 2021-06-05