Apache Beam DataFlow runner 抛出设置错误

Question

我们正在使用 Beam Python SDK 构建数据管道并尝试在 Dataflow 上运行，但出现以下错误，

A setup error was detected in beamapp-xxxxyyyy-0322102737-03220329-8a74-harness-lm6v. Please refer to the worker-startup log for detailed information.

但找不到详细的工作器启动日志。

我们尝试增加内存大小、工人数量等，但仍然遇到同样的错误。

这是我们使用的命令，

python run.py \
--project=xyz \
--runner=DataflowRunner \
--staging_location=gs://xyz/staging \
--temp_location=gs://xyz/temp \
--requirements_file=requirements.txt \
--worker_machine_type n1-standard-8 \
--num_workers 2

管道sn-p，

data = pipeline | "load data" >> beam.io.Read(    
    beam.io.BigQuerySource(query="SELECT * FROM abc_table LIMIT 100")
)

data | "filter data" >> beam.Filter(lambda x: x.get('column_name') == value)

上述管道只是从 BigQuery 加载数据并根据某些列值进行过滤。此管道在 DirectRunner 中的工作方式类似于魅力，但在 Dataflow 上却失败了。

我们是否有任何明显的设置错误？其他人得到同样的错误？我们可以使用一些帮助来解决问题。

更新：

我们的管道代码分布在多个文件中，因此我们创建了一个 python 包。我们通过传递 --setup_file 参数而不是 --requirements_file 解决了设置错误问题。

Answer 1

我们通过向数据流发送一组不同的参数解决了此设置错误问题。我们的代码分布在多个文件中，因此必须为其创建一个包。如果我们使用--requirements_file，作业将开始，但最终会失败，因为它无法在工人中找到包。 Beam Python SDK 有时不会为这些抛出明确的错误消息，它会重试作业并失败。要让您的代码与包一起运行，您需要传递 --setup_file 参数，其中列出了依赖项。确保python setup.py sdist 命令创建的包包含管道代码所需的所有文件。

如果您有一个私有托管的 python 包依赖项，则将 --extra_package 与 package.tar.gz 文件的路径一起传递。更好的方法是存储在 GCS 存储桶中并在此处传递路径。

我已经编写了一个示例项目来开始在 Dataflow 上使用 Apache Beam Python SDK - https://github.com/RajeshHegde/apache-beam-example

在这里阅读 - https://medium.com/@rajeshhegde/data-pipeline-using-apache-beam-python-sdk-on-dataflow-6bb8550bf366

Answer 2

我正在使用 Apache Beam/Dataflow 构建预测管道。我需要将模型文件包含在远程工作人员可用的依赖项中。 Dataflow 作业失败并显示相同的错误日志：

Error message from worker: A setup error was detected in beamapp-xxx-xxxxxxxxxx-xxxxxxxx-xxxx-harness-xxxx. Please refer to the worker-startup log for detailed information.

但是，此错误消息没有提供有关 worker-startup 日志的任何详细信息。最后，我找到了让工作人员登录并解决问题的方法。

众所周知，Dataflow 创建计算引擎来运行作业并在其上保存日志，以便我们可以访问 vm 以查看日志。我们可以通过 SSH 从 GCP 控制台连接到我们的 Dataflow 作业正在使用的 vm。然后我们可以查看位于/var/log/dataflow/taskrunner/harness的boot-json.log文件：

$ cd /var/log/dataflow/taskrunner/harness
$ cat boot-json.log

这里我们要注意了。在批处理模式下运行时，Dataflow 创建的 vm 是短暂的，并在作业失败时关闭。如果 vm 关闭了，我们就不能再访问它了。但是包含失败项的进程会重试 4 次，因此通常我们有足够的时间打开 boot-json.log 并查看发生了什么。

最后，我把我的 Python 设置解决方案放在这里，可能对其他人有帮助：

main.py

...
model_path = os.path.dirname(os.path.abspath(__file__)) + '/models/net.pd'
# pipeline code
...

MANIFEST.in

include models/*.*

setup.py complete example

REQUIRED_PACKAGES = [...]

setuptools.setup(
    ...
    include_package_data=True,
    install_requires=REQUIRED_PACKAGES,
    packages=setuptools.find_packages()
)

运行数据流管道

$ python main.py --setup_file=/absolute/path/to/setup.py ...