是否有一个包含多个文件的 Python 数据流 Flex 模板示例,其中脚本导入同一文件夹中包含的其他文件?
我的项目结构是这样的:
├── pipeline
│ ├── __init__.py
│ ├── main.py
│ ├── setup.py
│ ├── custom.py
我正在尝试将 custom.py 导入到 main.py 中作为数据流 Flex 模板。
我在管道执行中收到以下错误:
ModuleNotFoundError: No module named 'custom'
如果我将所有代码包含在一个文件中并且不进行任何导入,则管道工作正常。
示例 Dockerfile:
FROM gcr.io/dataflow-templates-base/python3-template-launcher-base
ARG WORKDIR=/dataflow/template/pipeline
RUN mkdir -p ${WORKDIR}
WORKDIR ${WORKDIR}
COPY pipeline /dataflow/template/pipeline
COPY spec/python_command_spec.json /dataflow/template/
ENV DATAFLOW_PYTHON_COMMAND_SPEC /dataflow/template/python_command_spec.json
RUN pip install avro-python3 pyarrow==0.11.1 apache-beam[gcp]==2.24.0
ENV FLEX_TEMPLATE_PYTHON_SETUP_FILE="${WORKDIR}/setup.py"
ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/main.py"
Python 规范文件:
{
"pyFile":"/dataflow/template/pipeline/main.py"
}
我正在使用以下命令部署模板:
gcloud builds submit --project=${PROJECT} --tag ${TARGET_GCR_IMAGE} .
我实际上通过向模板执行传递一个附加参数 setup_file 来解决这个问题。还需要在模板元数据中添加
setup_file
参数
--parameters setup_file="/dataflow/template/pipeline/setup.py"
显然 Dockerfile 中的命令
ENV FLEX_TEMPLATE_PYTHON_SETUP_FILE="${WORKDIR}/setup.py"
没有用,实际上并没有获取安装文件。
我的安装文件如下所示:
import setuptools
setuptools.setup(
packages=setuptools.find_packages(),
install_requires=[
'apache-beam[gcp]==2.24.0'
],
)
经过一些测试,我发现由于某些未知原因,工作目录(
WORKDIR
)中的 phyton 文件无法通过导入引用。但如果您创建一个子文件夹并将 python 依赖项移入其中,它就会起作用。我测试过并且它有效,例如,在您的用例中,您可以具有以下结构:
├── pipeline
│ ├── main.py
│ ├── setup.py
│ ├── mypackage
│ │ ├── __init__.py
│ │ ├── custom.py
您将可以参考:
import mypackage.custom
。 Docker 文件应在 custom.py
中移至正确的目录。
RUN mkdir -p ${WORKDIR}/mypackage
RUN touch ${WORKDIR}/mypackage/__init__.py
COPY custom.py ${WORKDIR}/mypackage
并且依赖会被添加到python安装目录中:
$ docker exec -it <container> /bin/bash
# find / -name custom.py
/usr/local/lib/python3.7/site-packages/mypackage/custom.py
对我来说,我不需要在命令中集成 setup_file 来触发 Flex 模板,这是我的 Dockerfile:
FROM gcr.io/dataflow-templates-base/python38-template-launcher-base
ARG WORKDIR=/dataflow/template
RUN mkdir -p ${WORKDIR}
WORKDIR ${WORKDIR}
COPY . .
ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/main.py"
ENV FLEX_TEMPLATE_PYTHON_SETUP_FILE="${WORKDIR}/setup.py"
# Install apache-beam and other dependencies to launch the pipeline
RUN pip install apache-beam[gcp]
RUN pip install -U -r ./requirements.txt
这是命令:
gcloud dataflow flex-template run "job_ft" --template-file-gcs-location "$TEMPLATE_PATH" --parameters paramA="valA" --region "europe-west1"
这是我的解决方案:
Dockerfile:
FROM gcr.io/dataflow-templates-base/python3-template-launcher-base:flex_templates_base_image_release_20210120_RC00
ARG WORKDIR=/dataflow/template
RUN mkdir -p ${WORKDIR}
WORKDIR ${WORKDIR}
COPY requirements.txt .
# Read https://stackoverflow.com/questions/65766066/can-i-make-flex-template-jobs-take-less-than-10-minutes-before-they-start-to-pro#comment116304237_65766066
# to understand why apache-beam is not being installed from requirements.txt
RUN pip install --no-cache-dir -U apache-beam==2.26.0
RUN pip install --no-cache-dir -U -r ./requirements.txt
COPY mymodule.py setup.py ./
COPY protoc_gen protoc_gen/
ENV FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE="${WORKDIR}/requirements.txt"
ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/mymodule.py"
ENV FLEX_TEMPLATE_PYTHON_SETUP_FILE="${WORKDIR}/setup.py"
这是我的 setup.py:
import setuptools
setuptools.setup(
packages=setuptools.find_packages(),
install_requires=[],
name="my df job modules",
)
我有一堆管道都在同一个存储库中,其中所有管道都需要使用
utils
目录中的包。
我的解决方案是在每个管道目录中添加一个符号链接到 utils 目录。我不需要在本地运行、使用 Dataflow Runner 运行或创建并运行经典模板。但是,有必要运行 Flex 模板。
pipelines/
├── pipeline_1
│ ├── pipeline_1_metadata
│ ├── pipeline_1.py
│ ├── bin
│ │ ├── build_flex_template_and_image.sh
│ │ ├── run_flex_template.sh
│ │ ├── ...
│ ├── README.md
│ └── utils -> ../utils # Added this, and it worked
├── pipeline_2
│ ├── pipeline_2_metadata
│ ├── pipeline_2.py
│ ├── bin
│ │ ├── build_flex_template_and_image.sh
│ │ ├── run_flex_template.sh
│ │ ├── ...
│ ├── README.md
│ └── utils -> ../utils # Added this, and it worked
├── # etc.
├── requirements.txt
├── setup.py
|── utils
├── bigquery_utils.py
├── dprint.py
├── gcs_file_utils.py
└── misc.py
我的
setup.py
:
import setuptools
setuptools.setup(
name="repo_name_here",
version="0.2",
install_requires=[], # Maybe upgrade Beam here?
packages=setuptools.find_namespace_packages(exclude=["*venv*"]),
)
从基本目录中,我使用 Google 提供的 Docker 映像进行构建:
gcloud dataflow flex-template build "${TEMPLATE_FILE}" \
--image-gcr-path "${REGION}-docker.pkg.dev/${PROJECT}/${ARTIFACT_REPO}/dataflow/pipeline-1:latest" \
--sdk-language "PYTHON" \
--flex-template-base-image "PYTHON3" \
--py-path "." \
--metadata-file "pipeline_1/pipeline_1_metadata" \
--env "FLEX_TEMPLATE_PYTHON_PY_FILE=pipeline_1/pipeline_1.py" \
--env "FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE=flex_requirements.txt" \
--env "FLEX_TEMPLATE_PYTHON_SETUP_FILE=setup.py"
这完全可行,但现在唯一的麻烦是我不能将默认的
requirements.txt
与默认的 Python 图像一起使用,因为我不知道如何在我的 venv
中安装正确版本的 Python 3.9 和相应地更新 requirements.txt
,因此我通过 flex_requirements.txt
生成 cut -d "=" -f 1 requirements.txt > flex_requirements.txt
并让基础镜像找出依赖关系。这就是疯狂。但如果我在接下来的几天内无法弄清楚,那将是另一个堆栈溢出问题。