本文介紹了如何使用BEAM的外部Kafka變換(本地)消費(fèi)消息的處理方法，對(duì)大家解決問題具有一定的參考價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)吧！
問題描述
我正在嘗試運(yùn)行一個(gè)應(yīng)用程序，該應(yīng)用程序使用Kafka生產(chǎn)者(Python客戶端)和一個(gè)阿帕奇光束管道，它(目前)只是通過將這些消息打印到STDOUT來使用它們。
我了解，將Kafka外部轉(zhuǎn)換與ApacheBEAM一起使用是一項(xiàng)跨語言的工作，因?yàn)樗{(diào)用Java外部服務(wù)。我遵循了following link’s選項(xiàng)1：
選項(xiàng)1：使用默認(rèn)擴(kuò)展服務(wù)
這是使用Python時(shí)推薦且最簡(jiǎn)單的設(shè)置選項(xiàng)
卡夫卡會(huì)變身。此選項(xiàng)僅適用于光束2.22.0和
稍后。此選項(xiàng)要求在運(yùn)行梁之前滿足以下前提條件
管道。在管道所在的計(jì)算機(jī)上安裝Java運(yùn)行時(shí)
構(gòu)造并確?！甁ava’命令可用。
我正在運(yùn)行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
我不完全確定我應(yīng)該使用哪個(gè)Runner，因?yàn)閜ortability documentation似乎指向Universal Local Runner，但我似乎找不到這個(gè)Runder
在documentation中。
以下是我正在嘗試實(shí)現(xiàn)的代碼示例：
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)
我啟動(dòng)了該模塊，但我并不完全理解它是如何工作的，因?yàn)樗坪跸螺d了一些Java構(gòu)件并啟動(dòng)了一個(gè)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
'
總結(jié)一下我的問題，您能解釋一下當(dāng)我啟動(dòng)腳本時(shí)發(fā)生了什么嗎？我應(yīng)該用哪一個(gè)跑步者來做這件事？我如何修復(fù)此問題？
Runner
我認(rèn)為Universal Runner位于apache_beam.runners.portability.portable_runner.
下
這篇關(guān)于如何使用BEAM的外部Kafka變換(本地)消費(fèi)消息的文章就介紹到這了，希望我們推薦的答案對(duì)大家有所幫助，
