本文介紹了如何使用BEAM的外部Kafka變換(本地)消費消息的處理方法，對大家解決問題具有一定的參考價值，需要的朋友們下面隨著小編來一起學習吧！
問題描述
我正在嘗試運行一個應用程序，該應用程序使用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變換(本地)消費消息的文章就介紹到這了，希望我們推薦的答案對大家有所幫助，
