Dockerized Apache Beam 返回“未提供 id”

Question

我在使用 dockerized Apache Beam 时遇到了问题。当尝试运行容器时，我只收到"No id provided."消息，仅此而已。这是代码和文件：

Dockerfile

FROM apache/beam_python3.8_sdk:latest
RUN apt update
RUN apt install -y wget curl unzip git
COPY ./ /root/data_analysis/
WORKDIR /root/data_analysis
RUN python3 -m pip install -r data_analysis/beam/requirements.txt
ENV PYTHONPATH=/root/data_analysis
ENV WORKER_ID=1
CMD python3 data_analysis/analysis.py

代码analysis.py：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

def run():
    options = PipelineOptions(["--runner=DirectRunner"])

    with beam.Pipeline(options=options) as p:
        p | beam.Create([1, 2, 3]) | beam.Map(lambda x: x-1) | beam.Map(print)

if __name__ == "__main__":
    run()

命令：

% docker build -f Dockerfile_beam -t beam .
[+] Building 242.2s (12/12) FINISHED                                                                                                                                                                                          
...

% docker run --name=beam beam   
2021/09/15 13:44:07 No id provided.

我发现此错误消息很可能是由这一行生成的：https://github.com/apache/beam/blob/410ad7699621e28433d81809f6b9c42fe7bd6a60/sdks/python/container/boot.go#L98

但是这是什么意思？这是哪个id？我缺少什么？

Answer 1

此错误很可能是由于您的 Docker 映像基于 SDK 工具映像 ( apache/beam_python3.8_sdk) 所致。SDK 线束图像用于可移植管道；当可移植运行程序需要执行必须以其原始语言执行的管道阶段时，它会使用 SDK 工具启动一个容器，并将管道该阶段的执行委托给 SDK 工具。因此，当 SDK 线束启动时，它期望获得启动它的运行器提供的各种配置详细信息，其中之一就是 ID。当您直接启动此容器时，不会提供这些配置详细信息，并且会崩溃。

对于您的特定用例的上下文，让我首先列出运行可移植管道所涉及的不同流程。

Pipeline Construction <---> Job Service <---> SDK Harness
                                         \--> Cross-Language SDK Harness

• 管道构建- 定义和运行管道的过程。它将您的管道定义发送到作业服务并接收管道结果。它不执行任何管道。
• 就业服务- 为您选择的跑步者提供的流程。这可能采用与原始管道构造不同的语言，因此无法运行用户代码，例如自定义 DoFns。
• SDK Harness - 执行用户代码的进程，由作业服务启动和管理。默认情况下，它位于 docker 容器中。
• 跨语言 SDK 框架 使用与管道构建不同的语言执行代码的进程。在您的例子中，Python 的 Kafka IO 使用跨语言，并且实际上在 Java SDK 工具中执行。

目前，您创建的 docker 容器基于 SDKharness 容器，这听起来不像您想要的。您似乎一直在尝试容器化您的管道构建代码​​，却意外地容器化了 SDK 工具。但既然您描述了您希望 ReadFromKafka 消费者被容器化，那么听起来您需要的是除了它使用的任何 SDK 工具之外，作业服务器也被容器化。

将作业服务器容器化是可能的，并且可能已经完成。例如，这是一个容器化的 Flink Job Server。容器化作业服务器可能会给您带来一些工件方面的麻烦，因为容器无法访问本地计算机上的工件暂存目录，但可能有一些方法可以解决这个问题。

此外，您提到您希望避免 SDK 线束在嵌套的 docker 容器中启动。如果您为 SDKharness 启动工作池 docker 容器并将其设置为外部环境，则运行程序（假设它支持外部环境）将尝试连接到您提供的 URL，而不是创建新的 docker 容器。如果可以在 Python SDK 中进行配置，您还需要针对 Java 跨语言环境进行配置。此配置应通过 python 的管道选项完成。--environment_type都是--environment_options很好的起点。