如何使用BEAM的外部Kafka变换(本地)消费消息 [英] How to consume messages using beam's external kafka transform (locally)

问题描述

我正在尝试运行一个应用程序，该应用程序使用Kafka生产者(Python客户端)和一个阿帕奇光束管道，它(目前)只是通过将这些消息打印到STDOUT来使用它们。

我了解，将Kafka外部转换与ApacheBEAM一起使用是一项跨语言的工作，因为它调用Java外部服务。我遵循了following link's选项1：

选项1：使用默认扩展服务

这是使用Python时推荐且最简单的设置选项 卡夫卡会变身。此选项仅适用于光束2.22.0和 稍后。

此选项要求在运行梁之前满足以下前提条件 管道。

在管道所在的计算机上安装Java运行时 构造并确保‘Java’命令可用。

我正在运行apache-beam==2.31.0，刚刚安装了Java：

openjdk 11.0.11 2021-04-20
OpenJDK Runtime Environment (build 11.0.11+9-Ubuntu-0ubuntu2.18.04)
OpenJDK 64-Bit Server VM (build 11.0.11+9-Ubuntu-0ubuntu2.18.04, mixed mode, sharing

我不完全确定我应该使用哪个Runner，因为portability documentation似乎指向Universal Local Runner，但我似乎找不到这个Runder 在documentation中。

以下是我正在尝试实现的代码示例：

import argparse
import apache_beam as beam
from helpers import ccloud_lib

from apache_beam.io.external.kafka import ReadFromKafka
from apache_beam.options.pipeline_options import PipelineOptions


def run(argv=None):
    """Main entry point; runs a word_count pipeline"""

    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input_topic",
        dest="input_topic",
        default="wordcount",
        help="Kafka topic to use for input",
    )
    parser.add_argument(
        "--kafka_config",
        dest="config_file",
        default="config/confluent/python.config",
    )

    args = parser.parse_known_args(argv)[0]
    beam_options = PipelineOptions(runner="DirectRunner")

    consumer_conf = ccloud_lib.read_ccloud_config(args.config_file)
    consumer_conf["group.id"] = "python_wordcount_group_1"
    consumer_conf["auto.offset.reset"] = "earliest"

    with beam.Pipeline(options=beam_options) as pipeline:
        pipeline
        | "Read"
        >> ReadFromKafka(
            consumer_config=consumer_conf,
            topics=[args.input_topic],
        )
        | "Print" >> beam.Map(print)

我启动了该模块，但我并不完全理解它是如何工作的，因为似乎下载了一些Java构件并启动了一个docker图像。然后我收到以下警告消息：

INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:b'2021/08/25 14:38:05 Failed to obtain provisioning information: failed to dial server at localhost:36071
	caused by:
context deadline exceeded
'

总结一下我的问题，您能解释一下当我启动脚本时发生了什么吗？我应该用哪一个跑步者来做这件事？我如何修复此问题？

Runner

我认为Universal Runner位于apache_beam.runners.portability.portable_runner.

下

这篇关于如何使用BEAM的外部Kafka变换(本地)消费消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！