企业官网建设流程全解析

• 大数据
• 批处理
• 流处理
• 数据工程

导读

本文围绕 Apache Beam Python SDK 内置的 YAML 流水线 DSL 及其 Jinja2 模板预处理机制，深入讲解如何使用% import指令将全部变换（transforms）与配置抽离到独立的宏文件（macros file）中，实现"一份主流水线 + 一份可复用宏库"的声明式开发模式。读完本文，你将掌握--jinja_variables传参、宏定义与调用、indent过滤器缩进等关键技术，并能在本地或 Dataflow 上直接运行仓库中现成的 WordCount 示例。

一、场景与价值：为什么用% import组织 Beam YAML 流水线

Apache Beam 的 Python SDK 支持以纯 YAML 声明式方式编写批处理与流处理流水线，入口为python -m apache_beam.yaml.main。当流水线规模变大、多个任务需要复用同一组输入输出约定时，把变换定义写死在每个 YAML 文件里会导致大量重复。

仓库中的 jinja/import 示例目录 正是为此设计的：它利用 Jinja2 的% import指令，让一个主流水线文件（wordCountImport.yaml）与一个宏文件（macros/wordCountMacros.yaml）协同工作——宏文件集中存放所有变换片段和配置模板，主文件只负责组装调用。这样每个变换的"标准写法"只有一份，改一处即全局生效，配置参数则通过命令行以 JSON 形式注入。

从源码结构看，该目录属于sdks/python/apache_beam/yaml/examples/transforms/jinja/，与include（% include子模块拼装）和inheritance（Jinja2 继承扩展）两个示例互为补充，共同构成 Beam YAML 模板化的三种组织范式。本示例聚焦% import的宏复用能力。

二、示例文件结构：主流水线与宏文件的分工

两个文件位于同一目录层级，宏文件被放在macros/子目录中，通过仓库根目录相对的 import 路径引用。

三、主流水线拆解：wordCountImport.yaml 的组装逻辑

主流水线文件的开头执行导入指令：

{% import 'apache_beam/yaml/examples/transforms/jinja/import/macros/wordCountMacros.yaml' as macros %}

这里import ... as macros把整个宏文件作为命名空间macros引入，之后即可用macros.宏名(...)调用。注意该路径是以 Beam 仓库根目录为基准的绝对相对路径，之所以能直接命中，是因为 Beam 的 Jinja2 环境把 beam 包根目录注册进了搜索路径（详见第五节"底层原理"）。

随后流水线主体是一个type: chain的变换链，六个步骤按顺序排列：

pipeline: type: chain transforms: # 第 1 步：读取文本文件 {{ macros.readFromTextTransform(readFromTextTransform) | indent(4, true) }} # 第 2 步：切分单词并计数 - name: Split words type: MapToFields config: {{ macros.mapToFieldsSplitConfig(mapToFieldsSplitConfig) | indent(8, true) }} # 第 3 步：拆成单个单词 {{ macros.explodeTransform(explodeTransform) | indent(4, true) }} # 第 4 步：按单词分组 {{ macros.combineTransform(combineTransform) | indent(4, true) }} # 第 5 步：格式化为 "word - count" - name: Format output type: MapToFields config: {{ macros.mapToFieldsCountConfig(mapToFieldsCountConfig) | indent(8, true) }} # 第 6 步：写出文本 {{ macros.writeToTextTransform(writeToTextTransform) | indent(4, true) }}

关键细节：

• 宏参数即渲染变量：macros.readFromTextTransform(readFromTextTransform)中的实参readFromTextTransform是 Jinja 变量名，其值由命令行的--jinja_variablesJSON 注入。
• indent(N, true)过滤器：宏输出是完整的多行 YAML 片段，必须通过| indent(4, true)统一缩进到父级正确的层级，第二个参数true表示首行也参与缩进。宏文件里第 2、5 步只输出config:内部的键值（不含- name与type行），因此外部补写这两行后用indent(8, true)对齐到config:之下——这是主文件与宏文件配合时的排版约定。

主流水线文件底部还给出了运行后的期望输出（真实数据来自莎士比亚《李尔王》全文）：

Row(output='king - 311') Row(output='lear - 253') Row(output='dramatis - 1') Row(output='personae - 1') Row(output='of - 483') Row(output='britain - 2') Row(output='france - 32') Row(output='duke - 26') Row(output='burgundy - 20') Row(output='cornwall - 75')

这组数据可直接用作验证流水线是否按预期运行的基准。

四、宏文件拆解：wordCountMacros.yaml 的六个宏

宏文件用{%- macro 宏名(params) -%}定义，{%-与-%}用于吞掉宏定义两侧多余的空白与换行，保证渲染输出干净。六个宏与主流水线步骤一一对应：

1.readFromTextTransform—— 读取 GCS 文本

{%- macro readFromTextTransform(params) -%} - name: Read from GCS type: ReadFromText config: path: "{{ params.path }}" {%- endmacro -%}

只暴露params.path一个参数，其余结构固定。ReadFromText是 Beam YAML 内置的标准 I/O 变换，此处读取的默认数据源为公共 GCS 文件gs://dataflow-samples/shakespeare/kinglear.txt。

2.mapToFieldsSplitConfig—— 切分与词形归一

{%- macro mapToFieldsSplitConfig(params) -%} language: "{{ params.language }}" fields: value: "{{ params.fields.value }}" word: callable: |- import re def my_mapping(row): return re.findall(r'[A-Za-z\']+', row.line.lower()) {%- endmacro -%}

该宏渲染的是MapToFields变换的config内容：language: python表示用 Python 表达式，value字段取值为常量"1"，word字段通过内联callable定义正则切词逻辑（转小写后提取字母与撇号）。这是本示例中唯一"业务逻辑内嵌"的位置，其余逻辑均以参数形式注入。

3.explodeTransform—— 数组展开

{%- macro explodeTransform(params) -%} - name: Explode word arrays type: Explode config: fields: "{{ params.fields }}" {%- endmacro -%}

Explode将word字段的单词列表展开为一行一个单词，参数fields: word指定展开目标字段。

4.combineTransform—— 分组计数

{%- macro combineTransform(params) -%} - name: Count words type: Combine config: group_by: "{{ params.group_by }}" combine: value: "{{ params.combine.value }}" {%- endmacro -%}

Combine按group_by: word分组，并对value字段执行sum聚合，得到每个单词的出现次数。

5.mapToFieldsCountConfig—— 输出格式化

{%- macro mapToFieldsCountConfig(params) -%} language: "{{ params.language }}" fields: output: '{{ params.fields.output }}' {%- endmacro -%}

output字段用 Python 表达式word + " - " + str(value)拼接成单词 - 次数的字符串，最终产生前文所示的Row(output='...')结果。

6.writeToTextTransform—— 写出结果

{%- macro writeToTextTransform(params) -%} - name: Write to GCS type: WriteToText config: path: "{{ params.path }}" {%- endmacro -%}

WriteToText把结果写到params.path指定的 GCS 路径（运行时替换为你的gs://MY-BUCKET/wordCounts/）。

五、运行准备：环境变量与数据源

按官方 README.md 的 General setup 完成准备：

export PIPELINE_FILE=apache_beam/yaml/examples/transforms/jinja/import/wordCountImport.yaml export KINGLEAR="gs://dataflow-samples/shakespeare/kinglear.txt" export TEMP_LOCATION="gs://MY-BUCKET/wordCounts/" export PROJECT="MY-PROJECT" export REGION="MY-REGION" cd <PATH_TO_BEAM_REPO>/beam/sdks/python

前提与注意事项：

• 仓库中的 wordCountImport.yaml 头部注释明确说明：默认读取的是 Google Cloud 公共文件，需要配置 Google Cloud 应用默认凭据（ADC）；若不想走 GCS，可把ReadFromText的path改为本地文件。
• 需要把MY-BUCKET、MY-PROJECT、MY-REGION替换成实际值。
• cd进入beam/sdks/python目录的目的是让apache_beam模块与仓库内的示例路径可直接被解析。

六、运行命令：多行与单行两种传参方式

--jinja_variables接受一个 JSON 字符串，其键名必须与主流水线中宏调用的实参名完全一致。官方给出两种写法：

多行写法

python -m apache_beam.yaml.main \ --project=${PROJECT} \ --region=${REGION} \ --yaml_pipeline_file="${PIPELINE_FILE}" \ --jinja_variables='{ "readFromTextTransform": {"path": "'"${KINGLEAR}"'"}, "mapToFieldsSplitConfig": { "language": "python", "fields": { "value": "1" } }, "explodeTransform": {"fields": "word"}, "combineTransform": { "group_by": "word", "combine": {"value": "sum"} }, "mapToFieldsCountConfig": { "language": "python", "fields": {"output": "word + \" - \" + str(value)"} }, "writeToTextTransform": {"path": "'"${TEMP_LOCATION}"'"} }'

单行写法

python -m apache_beam.yaml.main --project=${PROJECT} --region=${REGION} \ --yaml_pipeline_file="${PIPELINE_FILE}" --jinja_variables='{"readFromTextTransform": {"path": "'"${KINGLEAR}"'"}, "mapToFieldsSplitConfig": {"language": "python", "fields":{"value":"1"}}, "explodeTransform":{"fields":"word"}, "combineTransform":{"group_by":"word", "combine":{"value":"sum"}}, "mapToFieldsCountConfig":{"language": "python", "fields":{"output":"word + \" - \" + str(value)"}}, "writeToTextTransform":{"path":"'"${TEMP_LOCATION}"'"}}'

两种写法完全等价，单行形式更适合脚本化或模板化工具。

参数与宏的对应关系

采用这种"参数外置"的设计后，同一个宏文件可以服务于多个流水线：只要传参一致，渲染出的 YAML 就完全一致；传参变化时，只有对应变换的配置变化，结构保持不变。

七、底层原理：Beam 如何渲染 Jinja2 模板

1. 渲染入口与StrictUndefined严格模式

模板渲染的核心实现位于 yaml_transform.py 的expand_jinja函数：

def expand_jinja(jinja_template, jinja_variables, search_paths=()): beam_root_dir = os.path.dirname(os.path.dirname(os.path.abspath(beam.__file__))) all_search_paths = list(search_paths) if beam_root_dir not in all_search_paths: all_search_paths.append(beam_root_dir) if '.' not in all_search_paths: all_search_paths.append('.') return ( jinja2.Environment( undefined=jinja2.StrictUndefined, loader=_BeamFileIOLoader(all_search_paths)) .from_string(strip_leading_comments(jinja_template)) .render(datetime=datetime, **jinja_variables))

三个值得注意的设计：

• StrictUndefined严格模式：任何模板引用了未注入的变量都会直接抛错，而不是静默渲染成空串。这保证了--jinja_variables漏传参数时流水线会在启动阶段立即失败，避免带着残缺配置跑到运行时。
• beam 根目录自动加入搜索路径：这是{% import 'apache_beam/yaml/examples/transforms/jinja/import/macros/wordCountMacros.yaml' %}能按仓库相对路径解析的根本原因；同时.也被加入，支持以当前目录为基准的相对导入。
• 内置datetime命名空间：模板中可直接使用datetime.now()等函数生成时间戳，便于在输出路径中注入日期。

2. 自定义加载器_BeamFileIOLoader

_BeamFileIOLoader继承自jinja2.BaseLoader，把文件读取统一走 Beam 的FileSystems抽象层，因此 import/include 的模板文件既可以是本地文件，也可以是 GCS 等云存储对象。加载时还会通过strip_leading_comments剥离模板文件头部的 Apache License 注释，保证渲染产物是干净的 YAML。

3. 入口参数解析与--jinja_variable_flags

入口脚本 main.py 中，--jinja_variables以type=json.loads声明，即命令行参数本身就是 JSON 字典。此外还实现了_preparse_jinja_flags机制：通过--jinja_variable_flags声明一组标志名，把这些标志自动提升为 Jinja 变量并入--jinja_variables，目的是方便 Dataflow 模板等工具以扁平的--flag=value形式传参。若标志名与既有 PipelineOption 冲突，则跳过并强制走 JSON 方式（见 main.py）。

4. 测试侧对模板化的支持

仓库的示例测试 examples_test.py 展示了另一种渲染路径：测试数据源input_data.word_count_jinja_parameter_data()提供渲染所需的变量 JSON，测试框架用jinja2.DictLoader+StrictUndefined在内存中加载模板与宏文件后调用template.render(jinja_variables)。注释中说明标准expand_jinja暂不支持% include模板化，故测试采用DictLoader方案（对应仓库 TODO 编号 #35936），% import宏同样走这套测试渲染流程。这意味着本示例不仅可手动运行，也在仓库的 YAML 示例测试集中被持续验证。

八、运行验证与输出检查

流水线运行成功后，结果将写入TEMP_LOCATION指定的 GCS 目录。核验要点：

1. 输出目录下应生成分片文件（如wordCounts-00000-of-0000N），每行格式为单词 - 次数。
2. 与主流水线文件中的"Expected"注释对比抽样：king - 311、lear - 253、of - 483等高频词应稳定出现。
3. 若使用 Dataflow Runner 在云上运行，可在 Dataflow 控制台观察 Job 的 6 个变换步骤是否按Read from GCS → Split words → Explode word arrays → Count words → Format output → Write to GCS顺序执行。

九、与其他 Jinja 组织方式的对比与选型

Beam YAML 的 Jinja 能力不止% import一种用法，同目录下还有两个对照示例：

• include 示例：用% include把每个变换拆成独立子模块文件（submodules/下 6 个 yaml 文件），主文件按需引入。适合"变换片段粒度细、子模块各自独立演进"的场景。
• inheritance 示例：利用 Jinja2 继承（extends+block），基流水线base/base_pipeline.yaml预留extra_steps块，子流水线注入额外变换。适合"多版本流水线共享骨架、仅在特定位置插入/覆盖步骤"的场景。

三者对比：% import适合把"整套变换与配置"封装成命名空间复用（宏可带参数、可批量复用）；% include适合"按需拼接独立片段"；inheritance适合"骨架化、差异化的多分支流水线"。如果只是单个流水线、无复用诉求，则无需引入任何 Jinja 特性，直接写纯 YAML 即可。

十、实战注意事项

1. 严格传参：由于使用StrictUndefined，--jinja_variables中漏掉任何宏实参都会导致渲染报错；调试时留意报错信息中的变量名即可快速定位。
2. 注意缩进过滤器：宏输出多行片段时必须用| indent(N, true)对齐到目标层级，否则生成的 YAML 非法。改宏内部结构时，要同步核对主文件里的indent参数。
3. 路径基准：% import的路径以 beam 仓库根目录为基准（由expand_jinja自动注册搜索路径保证）；若把示例移到别处，需相应调整 import 路径或通过search_paths注入新基准。
4. 宏的注释与空白：宏定义用{%- ... -%}形态以消除多余换行；mapToFieldsSplitConfig中的内联callable用|-块标量保留多行 Python 源码，缩进必须保持 YAML 语义。
5. 云上运行：在 Dataflow Runner 下运行需追加--runner=DataflowRunner --temp_location等标准参数（参考 examples/README.md 的 Kafka 示例传参风格），并确保 ADC 凭据与 GCS 权限就绪。

• 大数据
• 批处理
• 流处理
• 数据工程