企业官网建设流程全解析

使用 Kestra 构建生产级 ETL 工作流：Data Engineering Zoomcamp 第二模块工作流编排实战指南

【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 👇🏼项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp

本指南围绕 Data Engineering Zoomcamp 课程第二模块（Workflow Orchestration）展开，完整讲解如何用开源编排平台 Kestra 以纯 YAML 的方式构建、调度、回填并部署数据管道。你将学会从 HTTP API 提取数据、经 Python 转换后用 DuckDB 查询的入门管道，到将纽约出租车（Yellow/Green Taxi）CSV 数据加载进本地 Postgres 与云端 GCS + BigQuery 的完整 ETL 流程，并掌握用 dbt 在 Kestra 内做数据转换、用 Schedule 触发器做定时调度与历史回填的实战技巧。

模块概览：为什么需要工作流编排

第二模块是 Data Engineering Zoomcamp 课程的核心转折点——从第一模块的"单机管道"走向"可编排、可调度、可回填"的工程化管道。本模块选用 Kestra 作为编排引擎，其核心理念是：

• 事件驱动：既支持基于时间的 Cron 调度，也支持基于事件的触发；
• 基础设施即代码（IaC）：用几行 YAML 声明式地描述整个工作流，流程可版本化、可审查、可复用；
• 开箱即用的插件生态：HTTP 下载、Python/SQL 脚本、JDBC、GCS、BigQuery、dbt、Git 同步等能力均以插件形式内置于任务类型中。

本模块的课程结构分为四大部分：

1. 概念部分：工作流编排与 Kestra 核心概念；
2. 动手实践：用 Kestra 为 NYC 出租车数据构建 ETL 管道（本地 Postgres 版）；
3. 云端实践：将同一套管道迁移到 GCS + BigQuery；
4. 选学拓展：将 Kestra 部署到云端并接入 Git 实现生产级工作流管理。

对应的全部流程文件位于仓库 cohorts/2025/02-workflow-orchestration/flows 目录，共 9 个 YAML 文件，是本文所有示例的源码出处。

环境准备：用 Docker Compose 快速安装 Kestra

启动 Kestra 服务

本模块推荐使用 Docker Compose 一次性拉起两个容器：Kestra 服务器 + 其专属的 Postgres 元数据库（Kestra 用它存储流程定义、执行记录与队列）。仓库根目录提供了可直接使用的编排文件 02-workflow-orchestration/docker-compose.yml，其中同时包含练习用的pgdatabase（Postgres，端口 5432）、pgadmin（端口 8085）、kestra_postgres（Kestra 元数据库）与kestra四个服务。

cd 02-workflow-orchestration docker compose up -d

容器启动后，在浏览器访问 http://localhost:8080 即可打开 Kestra 的 Web UI。

版本要求（重要）：本模块要求 Postgres 镜像使用PostgreSQL 15 或更高版本（推荐 latest）。原因在于加载出租车数据的流程使用MERGE语句实现幂等 upsert，而MERGE是 PostgreSQL 15 才引入的语法，旧版本会直接报语法错误。

通过 API 批量导入流程

除了在 UI 中逐个粘贴 YAML 创建流程，也可以使用 Kestra 的 REST API 程序化导入（注意路径前缀要与执行目录一致）：

curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/01_getting_started_data_pipeline.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/02_postgres_taxi.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/02_postgres_taxi_scheduled.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/03_postgres_dbt.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/04_gcp_kv.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/05_gcp_setup.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/06_gcp_taxi.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/06_gcp_taxi_scheduled.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUpload=@flows/07_gcp_dbt.yaml

端口冲突提示：如果本机 8080 端口已被 pgAdmin 等其他程序占用，修改 docker-compose 中 Kestra 的端口映射（例如18080:8080），然后改用 http://localhost:18080 访问即可。

入门管道：HTTP 提取 → Python 转换 → DuckDB 查询

首先从最简单的一条流程看起，理解 Kestra 的"任务（task）"与"任务间数据传递"模型。该流程通过 HTTP REST API 提取数据、用 Python 做转换，再用 DuckDB 做聚合查询，完整代码见 01_getting_started_data_pipeline.yaml。

id: 01_getting_started_data_pipeline namespace: zoomcamp inputs: - id: columns_to_keep type: ARRAY itemType: STRING defaults: - brand - price tasks: - id: extract type: io.kestra.plugin.core.http.Download uri: https://dummyjson.com/products - id: transform type: io.kestra.plugin.scripts.python.Script containerImage: python:3.11-alpine inputFiles: data.json: "{{outputs.extract.uri}}" outputFiles: - "*.json" env: COLUMNS_TO_KEEP: "{{inputs.columns_to_keep}}" script: | import json import os columns_to_keep_str = os.getenv("COLUMNS_TO_KEEP") columns_to_keep = json.loads(columns_to_keep_str) with open("data.json", "r") as file: data = json.load(file) filtered_data = [ {column: product.get(column, "N/A") for column in columns_to_keep} for product in data["products"] ] with open("products.json", "w") as file: json.dump(filtered_data, file, indent=4) - id: query type: io.kestra.plugin.jdbc.duckdb.Query inputFiles: products.json: "{{outputs.transform.outputFiles['products.json']}}" sql: | INSTALL json; LOAD json; SELECT brand, round(avg(price), 2) as avg_price FROM read_json_auto('{{workingDir}}/products.json') GROUP BY brand ORDER BY avg_price DESC; fetchType: STORE

这条流程展示了 Kestra 的三个关键机制：

1. Inputs（流程入参）：columns_to_keep声明为ARRAY类型，运行时可在 UI 上修改默认值，也可通过{{inputs.columns_to_keep}}在任意任务中引用；
2. 任务间数据传递：extract任务（io.kestra.plugin.core.http.Download）下载的 JSON 文件通过{{outputs.extract.uri}}传给 Python 任务作为输入；Python 任务产出的products.json再通过{{outputs.transform.outputFiles['products.json']}}传给 DuckDB 任务，整个过程由 Kestra 内部存储（Internal Storage）托管；
3. 隔离执行环境：Python 脚本运行在python:3.11-alpine容器内，通过env注入参数，脚本无需硬编码任何配置。

在 UI 中运行该流程后，可切换到Gantt标签页查看各任务的时间线，在Logs标签页查看每个任务的运行日志，这是后续排查所有流程问题的基础操作。

本地 Postgres 管道：加载纽约出租车数据

数据源说明

课程使用的纽约市出租车数据来自 NYC Taxi & Limousine Commission (TLC)，但特别注意：官方 nyc.gov 网站目前只提供 Parquet 格式，而本课程刻意选用CSV 版本（托管在 DataTalksClub 的 release 中），目的是让初学者能用 Excel、Google Sheets 甚至文本编辑器直接检视数据，降低上手门槛。

手工触发版流程

核心流程文件为 02_postgres_taxi.yaml，它按"选择年月 → 打标签 → 提取 CSV → 建表 → 装载 → 合并"的链路运行：

入参与变量定义

inputs: - id: taxi type: SELECT displayName: Select taxi type values: [yellow, green] defaults: yellow - id: year type: SELECT displayName: Select year values: ["2019", "2020"] defaults: "2019" - id: month type: SELECT displayName: Select month values: ["01", "02", "03", "04", "05", "06", "07", "08", "09", "10", "11", "12"] defaults: "01" variables: file: "{{inputs.taxi}}_tripdata_{{inputs.year}}-{{inputs.month}}.csv" staging_table: "public.{{inputs.taxi}}_tripdata_staging" table: "public.{{inputs.taxi}}_tripdata" data: "{{outputs.extract.outputFiles[inputs.taxi ~ '_tripdata_' ~ inputs.year ~ '-' ~ inputs.month ~ '.csv']}}"

variables中的file用于拼接下载地址（如green_tripdata_2019-01.csv），data则动态指向extract任务的输出文件。注意data表达式里使用了 Pebble 模板的~字符串拼接运算符，这在整个模块的流程中随处可见。

提取任务：wget + gunzip

- id: extract type: io.kestra.plugin.scripts.shell.Commands outputFiles: - "*.csv" taskRunner: type: io.kestra.plugin.core.runner.Process commands: - wget -qO- https://github.com/DataTalksClub/nyc-tlc-data/releases/download/{{inputs.taxi}}/{{render(vars.file)}}.gz | gunzip > {{render(vars.file)}}

数据源是 gzip 压缩的 CSV，因此这里用wget -qO-将内容直接管道给gunzip解压为本地 CSV。outputFiles: ["*.csv"]声明该任务会产出 CSV 产物，供后续{{outputs.extract.outputFiles[...]}}引用。

建表与装载：IF 分支 + CopyIn

流程用io.kestra.plugin.core.flow.If按taxi输入分流，Yellow 与 Green 两套逻辑各自独立。以 Yellow 分支为例，其链路为：

- id: if_yellow_taxi type: io.kestra.plugin.core.flow.If condition: "{{inputs.taxi == 'yellow'}}" then: - id: yellow_create_table type: io.kestra.plugin.jdbc.postgresql.Queries sql: | CREATE TABLE IF NOT EXISTS {{render(vars.table)}} ( unique_row_id text, filename text, VendorID text, tpep_pickup_datetime timestamp, tpep_dropoff_datetime timestamp, passenger_count integer, trip_distance double precision, RatecodeID text, store_and_fwd_flag text, PULocationID text, DOLocationID text, payment_type integer, fare_amount double precision, extra double precision, mta_tax double precision, tip_amount double precision, tolls_amount double precision, improvement_surcharge double precision, total_amount double precision, congestion_surcharge double precision ); # 随后依次是 yellow_create_staging_table、yellow_truncate_staging_table、 # yellow_copy_in_to_staging_table（COPY CSV）、yellow_add_unique_id_and_filename、 # yellow_merge_data，详见源码文件

装载环节使用的是io.kestra.plugin.jdbc.postgresql.CopyIn任务，它对应 Postgres 的高效COPY协议，逐月数据写入staging表：

- id: yellow_copy_in_to_staging_table type: io.kestra.plugin.jdbc.postgresql.CopyIn format: CSV from: "{{render(vars.data)}}" table: "{{render(vars.staging_table)}}" header: true columns: [VendorID,tpep_pickup_datetime,tpep_dropoff_datetime,passenger_count,trip_distance,RatecodeID,store_and_fwd_flag,PULocationID,DOLocationID,payment_type,fare_amount,extra,mta_tax,tip_amount,tolls_amount,improvement_surcharge,total_amount,congestion_surcharge]

幂等去重与合并：MD5 唯一键 + MERGE

由于不同月份的数据可能出现重复行，流程先为每行生成幂等唯一键（对 VendorID、上下车时间、上下车区域、费用、里程等关键字段拼接后取md5），再写入unique_row_id与来源filename：

UPDATE public.yellow_tripdata_staging SET unique_row_id = md5( COALESCE(CAST(VendorID AS text), '') || COALESCE(CAST(tpep_pickup_datetime AS text), '') || COALESCE(CAST(tpep_dropoff_datetime AS text), '') || COALESCE(PULocationID, '') || COALESCE(DOLocationID, '') || COALESCE(CAST(fare_amount AS text), '') || COALESCE(CAST(trip_distance AS text), '') ), filename = '{{render(vars.file)}}';

最后用 PostgreSQL 15 的MERGE语句，以unique_row_id为连接键把 staging 数据幂等合并进最终表：

MERGE INTO public.yellow_tripdata AS T USING public.yellow_tripdata_staging AS S ON T.unique_row_id = S.unique_row_id WHEN NOT MATCHED THEN INSERT (...) VALUES (...);

Green 分支结构与 Yellow 完全一致，区别仅在于时间字段为lpep_pickup_datetime/lpep_dropoff_datetime，且多了ehail_fee与trip_type两个字段。

数据库连接默认值：pluginDefaults

流程末尾通过pluginDefaults统一为所有io.kestra.plugin.jdbc.postgresql插件任务注入连接参数，避免每个任务重复书写：

pluginDefaults: - type: io.kestra.plugin.jdbc.postgresql values: url: jdbc:postgresql://host.docker.internal:5432/postgres-zoomcamp username: kestra password: k3str4

macOS/Windows 用户：这里的host.docker.internal可以直接访问宿主机端口映射出来的 Postgres；Linux 用户则需要注意host.docker.internal默认不可用（详见文末"常见问题排查"章节的联合 Compose 方案）。

最后一步purge_files（io.kestra.plugin.core.storage.PurgeCurrentExecutionFiles）会清理本次执行的中间产物，避免占用 Kestra 内部存储；若想保留产物便于调试，可在 UI 中禁用该任务。

调度与回填：让管道按 Cron 自动运行

手工触发版解决了"如何跑一次"，调度版 02_postgres_taxi_scheduled.yaml 则解决"如何定时跑、如何补历史数据"。

从 input 到 trigger.date

调度版不再让用户手工选年月，而是把文件名的年月部分替换为trigger.date（触发日期），每次调度执行时自动推算：

variables: file: "{{inputs.taxi}}_tripdata_{{trigger.date | date('yyyy-MM')}}.csv" data: "{{outputs.extract.outputFiles[inputs.taxi ~ '_tripdata_' ~ (trigger.date | date('yyyy-MM')) ~ '.csv']}}"

Schedule 触发器与 Cron

triggers: - id: green_schedule type: io.kestra.plugin.core.trigger.Schedule cron: "0 9 1 * *" inputs: taxi: green - id: yellow_schedule type: io.kestra.plugin.core.trigger.Schedule cron: "0 10 1 * *" inputs: taxi: yellow

• 0 9 1 * *表示每月 1 日 09:00 UTC运行 Green 数据的管道；0 10 1 * *表示同一时刻（10:00 UTC）运行 Yellow；
• 通过inputs.taxi把不同 taxi 类型注入同一条流程，实现"一个流程、多个调度"；
• 流程还通过concurrency: { limit: 1 }限制同一时刻只允许一个执行实例，避免回填与定时执行重叠冲突。

用 Backfill 补历史数据

对于按月分片的数据管道，Kestra 的Backfill（回填）功能会以触发时间为基准，为设定时间范围内的每个时间点各生成一次执行。操作方式：在 UI 中打开已调度的流程 → 选择 Backfill → 设定起止时间范围。由于数据集较大，本模块建议只回填 2019 年全年的 Green 数据作为练习。

最佳实践：在 UI 中为回填产生的执行手动添加backfill: true标签（流程注释中也有此提示），这样在列表里可以清晰区分"定时执行"与"回填执行"。

在 Kestra 内编排 dbt 模型（选学）

数据进入 Postgres 后，可以用 dbt 完成转换建模。流程 03_postgres_dbt.yaml 展示了 Kestra 编排 dbt 的标准姿势：

inputs: - id: dbt_command type: SELECT allowCustomValue: true defaults: dbt build values: - dbt build - dbt debug # 首次运行时先用它验证数据库连接 tasks: - id: sync type: io.kestra.plugin.git.SyncNamespaceFiles url: https://github.com/DataTalksClub/data-engineering-zoomcamp branch: main namespace: "{{ flow.namespace }}" gitDirectory: 04-analytics-engineering/taxi_rides_ny dryRun: false - id: dbt-build type: io.kestra.plugin.dbt.cli.DbtCLI env: DBT_DATABASE: postgres-zoomcamp DBT_SCHEMA: public namespaceFiles: enabled: true containerImage: ghcr.io/kestra-io/dbt-postgres:latest taskRunner: type: io.kestra.plugin.scripts.runner.docker.Docker networkMode: host commands: - dbt deps - "{{ inputs.dbt_command }}" storeManifest: key: manifest.json namespace: "{{ flow.namespace }}" profiles: | default: outputs: dev: type: postgres host: host.docker.internal user: kestra password: k3str4 port: 5432 dbname: postgres-zoomcamp schema: public threads: 8 connect_timeout: 10 priority: interactive target: dev

要点拆解：

1. Git 同步：SyncNamespaceFiles从课程仓库的04-analytics-engineering/taxi_rides_ny目录把 dbt 工程（models、macros、packages.yml 等）同步到 Kestra 命名空间。首次运行后可将该任务disabled: true省去重复拉取；
2. dbt CLI 容器化：DbtCLI任务运行在ghcr.io/kestra-io/dbt-postgres镜像中，dbt deps先安装依赖包（见 04-analytics-engineering/taxi_rides_ny/packages.yml），再执行dbt build；
3. profiles 内联：dbt 连接配置以profiles.yml形式直接内联在流程中，通过环境变量DBT_DATABASE/DBT_SCHEMA覆盖工程默认值；
4. Manifest 存储：storeManifest将 dbt 构建产物manifest.json存入命名空间，便于 Kestra 展示 lineage 信息。

本小节为选学内容，仅为作业提供铺垫；dbt 的完整讲解在课程的 04-analytics-engineering 模块。

云端管道：GCS 数据湖 + BigQuery 数仓

本地管道跑通后，将其迁移到 Google Cloud Platform（GCP）：用GCS 作为数据湖存放原始 CSV，用BigQuery 作为数据仓库做查询与分析。

第一步：用 KV Store 配置 GCP 凭据

流程 04_gcp_kv.yaml 用io.kestra.plugin.core.kv.Set任务把以下五个配置写入 Kestra 的 Key-Value Store（KV Store），供所有云端流程通过{{kv('KEY')}}引用：

安全警告：GCP_CREDS服务账号凭据属于敏感信息，务必像保管密码一样对待，绝不能提交到 Git。正式场景下更推荐用 Kestra 的 Secrets 机制存储敏感值，而把非敏感配置放在 KV Store，做到"敏感数据与流程逻辑分离"。

第二步：创建 GCS 桶与 BigQuery 数据集

如果第一模块尚未创建过这些资源，运行 05_gcp_setup.yaml 一键创建：

tasks: - id: create_gcs_bucket type: io.kestra.plugin.gcp.gcs.CreateBucket ifExists: SKIP storageClass: REGIONAL name: "{{kv('GCP_BUCKET_NAME')}}" - id: create_bq_dataset type: io.kestra.plugin.gcp.bigquery.CreateDataset name: "{{kv('GCP_DATASET')}}" ifExists: SKIP pluginDefaults: - type: io.kestra.plugin.gcp values: serviceAccount: "{{kv('GCP_CREDS')}}" projectId: "{{kv('GCP_PROJECT_ID')}}" location: "{{kv('GCP_LOCATION')}}" bucket: "{{kv('GCP_BUCKET_NAME')}}"

ifExists: SKIP保证资源已存在时不会报错；pluginDefaults让后续所有 GCP 插件任务自动继承凭据、项目、地域与桶配置。

第三步：加载出租车数据到 BigQuery

核心流程 06_gcp_taxi.yaml 的链路比本地版多了"上传 GCS"与"外部表"两步：

上传 GCS

variables: file: "{{inputs.taxi}}_tripdata_{{inputs.year}}-{{inputs.month}}.csv" gcs_file: "gs://{{kv('GCP_BUCKET_NAME')}}/{{vars.file}}" table: "{{kv('GCP_DATASET')}}.{{inputs.taxi}}_tripdata_{{inputs.year}}_{{inputs.month}}" tasks: - id: upload_to_gcs type: io.kestra.plugin.gcp.gcs.Upload from: "{{render(vars.data)}}" to: "{{render(vars.gcs_file)}}"

创建按日分区的主表

主表yellow_tripdata通过PARTITION BY DATE(tpep_pickup_datetime)按上车日期做日级分区，这是后续按时间过滤查询时控制成本的关键：

CREATE TABLE IF NOT EXISTS `{{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.yellow_tripdata` ( unique_row_id BYTES, filename STRING, VendorID STRING, tpep_pickup_datetime TIMESTAMP, ... congestion_surcharge NUMERIC ) PARTITION BY DATE(tpep_pickup_datetime);

外部表 → 临时表 → MERGE

装载采用"四步法"：先用CREATE OR REPLACE EXTERNAL TABLE将 GCS 中的 CSV 直接暴露为外部表（OPTIONS中指定format='CSV'、uris、skip_leading_rows=1、ignore_unknown_values=TRUE）；再用CREATE OR REPLACE TABLE ... AS SELECT生成带 MD5 唯一键的月度临时表；最后MERGE进主表实现幂等合并：

MERGE INTO `{{kv('GCP_PROJECT_ID')}}.{{kv('GCP_DATASET')}}.yellow_tripdata` T USING `{{kv('GCP_PROJECT_ID')}}.{{render(vars.table)}}` S ON T.unique_row_id = S.unique_row_id WHEN NOT MATCHED THEN INSERT (...) VALUES (...);

值得注意的是，BigQuery 版的unique_row_id只用 5 个字段（VendorID、pickup/dropoff 时间、PULocationID、DOLocationID）拼接取 MD5，与 Postgres 版略有差异，这也说明了"幂等键的设计取决于业务上如何定义重复行"。

第四步：调度 + 回填全量数据

调度版 06_gcp_taxi_scheduled.yaml 与本地版一致：通过trigger.date自动推算文件名与表名，用两个 Schedule 触发器分别驱动 Green（每月 1 日 09:00 UTC）与 Yellow（每月 1 日 10:00 UTC）。

由于云端存储与计算近乎无限可扩展，可以放心地回填 Yellow 与 Green 的全量历史数据，而无需担心本地机器资源耗尽——这正是"本地先小规模验证、云端全量回填"的经典迁移路径。

第五步：云端 dbt 转换（选学）

07_gcp_dbt.yaml 与 Postgres 版同构，差异在于：镜像换为ghcr.io/kestra-io/dbt-bigquery，服务账号 JSON 通过inputFiles: { sa.json: "{{kv('GCP_CREDS')}}" }注入容器，profiles 使用 BigQuery 适配器：

profiles: | default: outputs: dev: type: bigquery dataset: "{{kv('GCP_DATASET')}}" project: "{{kv('GCP_PROJECT_ID')}}" location: "{{kv('GCP_LOCATION')}}" keyfile: sa.json method: service-account priority: interactive threads: 16 timeout_seconds: 300 fixed_retries: 1 target: dev

注意：运行 dbt 流程前，可能需要先在 UI 中编辑同步过来的models/staging/schema.yml，把sources的database/schema调整为你的项目与数据集（Postgres 版为postgres-zoomcamp/public，BigQuery 版为项目 ID/zoomcamp）。

选学进阶：部署 Kestra 到云端并用 Git 管理

当管道在本机与云端都稳定运行后，可以把 Kestra 本身部署到 Google Cloud 生产环境，让它按已配置的调度持续运行，并通过 Git 仓库自动同步与部署工作流。

在生产化时有两点必须注意：

1. 敏感信息治理：工作流 YAML 中绝不能出现明文密码或凭据，应统一使用Secrets（加密存储）与KV Store（键值存储）来保存，实现"配置与代码分离"；
2. 版本化流程定义：将流程 YAML 提交到 Git，配合 CI/CD（如 GitHub Actions）实现流程的审查、测试与自动部署，这正是"基础设施即代码"在数据编排领域的落地。

常见问题排查

版本与端口

• 镜像选择：Kestra 应使用kestra/kestra:latest（最新稳定版）；不要使用kestra/kestra:develop（开发版，可能包含未修复的 bug）；
• Postgres 版本：必须 ≥ 15（MERGE语句依赖），直接使用postgres:latest；
• 端口冲突：若 pgAdmin 或其他程序占用 8080，把 Kestra 端口映射改为如18080:8080，再用 http://localhost:18080 访问。

Linux 下的 Connection Refused

在 Linux 上从 Kestra 容器内访问宿主机的 Postgres 会遇到Connection Refused，因为host.docker.internal在 Linux 上行为不同。解决办法是使用"全家桶" Docker Compose——把 Kestra、Kestra 元数据库、练习用 Postgres（postgres_zoomcamp）与 pgAdmin 放在同一个 Compose 文件中，Kestra 内部通过容器名postgres_zoomcamp而不是host.docker.internal访问数据库（pluginDefaults中相应修改）。该 Compose 的要点包括：

• 三个 named volume：postgres-data（Kestra 元数据）、kestra-data（Kestra 内部存储）、zoomcamp-data（练习数据）；
• kestra服务以user: "root"运行（仅为访问 Docker socket 的开发态做法），挂载docker.sock与/tmp/kestra-wd；
• postgres_zoomcamp提供练习库postgres-zoomcamp，端口5432:5432；
• pgadmin运行在 8085 端口，避开 Kestra 的 8080/8081。

完整 YAML 见原文档的 "Docker Compose Example" 折叠块。如果仍无法解决，请停止并移除现有 Kestra + Postgres 容器后重新docker-compose up -d。

BigQuery CSV 列数不匹配错误

如果遇到如下报错：

BigQueryError{reason=invalid, location=null, message=Error while reading table: kestra-sandbox.zooomcamp.yellow_tripdata_2020_01, error message: CSV table references column position 17, but line contains only 14 columns.; line_number: 2103925 byte_offset_to_start_of_line: 194863028 column_index: 17 column_name: "congestion_surcharge" column_type: NUMERIC File: gs://anna-geller/yellow_tripdata_2020-01.csv}

这通常不是 schema 问题，而是 CSV 文件在下载或上传过程中损坏（网络中断导致文件不完整），造成源表与目标表列数不一致。解决办法：重跑整个执行流程，强制重新下载 CSV 并重新上传到 GCS 即可。

模块作业与进阶挑战

本模块作业详见 cohorts/2025/02-workflow-orchestration/homework.md，核心任务是把现有流程扩展到2021 年数据（2021-01 至 2021-07），官方给出了两条路径：

1. 利用回填：在 06_gcp_taxi_scheduled.yaml 上执行 Backfill，时间范围设为2021-01-01至2021-07-31，并分别对yellow与green各执行一次；
2. 手动循环：用ForEach任务遍历"年月 × taxi 类型"组合，并通过Subflow子流程任务触发主流程，体会 Kestra 流程编排的复用能力。

作业还包含 6 道测验题，覆盖渲染变量值推断（如green_tripdata_2020-04.csv）、各年度行数统计、文件解压大小以及 Schedule 触发器时区配置（正确答案是timezone属性设为America/New_York）。

进一步探索

• 全部 9 个流程源码：cohorts/2025/02-workflow-orchestration/flows
• 本地一键环境：02-workflow-orchestration/docker-compose.yml
• dbt 工程示例：04-analytics-engineering/taxi_rides_ny
• 第一模块（Docker + Terraform）环境准备：01-docker-terraform
• 往届学员笔记与视频：可参考 2022 届、2023 届、2024 届 的社区笔记沉淀

完成本模块后，你将具备"用声明式 YAML 编排一条可调度、可回填、可上云的完整 ETL 管道"的核心能力，这套方法论将贯穿课程后续的数据仓库、dbt 与批处理模块。

【免费下载链接】data-engineering-zoomcampData Engineering Zoomcamp is a free 9-week course on building production-ready data pipelines. Join the course here 👇🏼项目地址: https://gitcode.com/GitHub_Trending/da/data-engineering-zoomcamp