使用 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 TaxiCSV 数据加载进本地 Postgres 与云端 GCS BigQuery 的完整 ETL 流程并掌握用 dbt 在 Kestra 内做数据转换、用 Schedule 触发器做定时调度与历史回填的实战技巧。模块概览为什么需要工作流编排第二模块是 Data Engineering Zoomcamp 课程的核心转折点——从第一模块的单机管道走向可编排、可调度、可回填的工程化管道。本模块选用 Kestra 作为编排引擎其核心理念是事件驱动既支持基于时间的 Cron 调度也支持基于事件的触发基础设施即代码IaC用几行 YAML 声明式地描述整个工作流流程可版本化、可审查、可复用开箱即用的插件生态HTTP 下载、Python/SQL 脚本、JDBC、GCS、BigQuery、dbt、Git 同步等能力均以插件形式内置于任务类型中。本模块的课程结构分为四大部分概念部分工作流编排与 Kestra 核心概念动手实践用 Kestra 为 NYC 出租车数据构建 ETL 管道本地 Postgres 版云端实践将同一套管道迁移到 GCS BigQuery选学拓展将 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其中同时包含练习用的pgdatabasePostgres端口 5432、pgadmin端口 8085、kestra_postgresKestra 元数据库与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 fileUploadflows/01_getting_started_data_pipeline.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/02_postgres_taxi.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/02_postgres_taxi_scheduled.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/03_postgres_dbt.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/04_gcp_kv.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/05_gcp_setup.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/06_gcp_taxi.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/06_gcp_taxi_scheduled.yaml curl -X POST http://localhost:8080/api/v1/flows/import -F fileUploadflows/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, indent4) - 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 的三个关键机制Inputs流程入参columns_to_keep声明为ARRAY类型运行时可在 UI 上修改默认值也可通过{{inputs.columns_to_keep}}在任意任务中引用任务间数据传递extract任务io.kestra.plugin.core.http.Download下载的 JSON 文件通过{{outputs.extract.uri}}传给 Python 任务作为输入Python 任务产出的products.json再通过{{outputs.transform.outputFiles[products.json]}}传给 DuckDB 任务整个过程由 Kestra 内部存储Internal Storage托管隔离执行环境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.csvdata则动态指向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_tableCOPY 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与来源filenameUPDATE 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: k3str4macOS/Windows 用户这里的host.docker.internal可以直接访问宿主机端口映射出来的 PostgresLinux 用户则需要注意host.docker.internal默认不可用详见文末常见问题排查章节的联合 Compose 方案。最后一步purge_filesio.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 触发器与 Crontriggers: - 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: yellow0 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要点拆解Git 同步SyncNamespaceFiles从课程仓库的04-analytics-engineering/taxi_rides_ny目录把 dbt 工程models、macros、packages.yml 等同步到 Kestra 命名空间。首次运行后可将该任务disabled: true省去重复拉取dbt CLI 容器化DbtCLI任务运行在ghcr.io/kestra-io/dbt-postgres镜像中dbt deps先安装依赖包见 04-analytics-engineering/taxi_rides_ny/packages.yml再执行dbt buildprofiles 内联dbt 连接配置以profiles.yml形式直接内联在流程中通过环境变量DBT_DATABASE/DBT_SCHEMA覆盖工程默认值Manifest 存储storeManifest将 dbt 构建产物manifest.json存入命名空间便于 Kestra 展示 lineage 信息。本小节为选学内容仅为作业提供铺垫dbt 的完整讲解在课程的 04-analytics-engineering 模块。云端管道GCS 数据湖 BigQuery 数仓本地管道跑通后将其迁移到 Google Cloud PlatformGCP用GCS 作为数据湖存放原始 CSV用BigQuery 作为数据仓库做查询与分析。第一步用 KV Store 配置 GCP 凭据流程 04_gcp_kv.yaml 用io.kestra.plugin.core.kv.Set任务把以下五个配置写入 Kestra 的 Key-Value StoreKV Store供所有云端流程通过{{kv(KEY)}}引用KV 键含义示例值GCP_CREDS服务账号 JSON 内容需替换为自己的 SA 凭据GCP_PROJECT_IDGCP 项目 IDkestra-sandboxGCP_LOCATION资源所在地域europe-west2GCP_BUCKET_NAMEGCS 桶名必须全局唯一your-name-kestraGCP_DATASETBigQuery 数据集名zoomcamp安全警告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与外部表两步上传 GCSvariables: 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中指定formatCSV、uris、skip_leading_rows1、ignore_unknown_valuesTRUE再用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/publicBigQuery 版为项目 ID/zoomcamp。选学进阶部署 Kestra 到云端并用 Git 管理当管道在本机与云端都稳定运行后可以把 Kestra 本身部署到 Google Cloud 生产环境让它按已配置的调度持续运行并通过 Git 仓库自动同步与部署工作流。在生产化时有两点必须注意敏感信息治理工作流 YAML 中绝不能出现明文密码或凭据应统一使用Secrets加密存储与KV Store键值存储来保存实现配置与代码分离版本化流程定义将流程 YAML 提交到 Git配合 CI/CD如 GitHub Actions实现流程的审查、测试与自动部署这正是基础设施即代码在数据编排领域的落地。常见问题排查版本与端口镜像选择Kestra 应使用kestra/kestra:latest最新稳定版不要使用kestra/kestra:develop开发版可能包含未修复的 bugPostgres 版本必须 ≥ 15MERGE语句依赖直接使用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 元数据库、练习用 Postgrespostgres_zoomcamp与 pgAdmin 放在同一个 Compose 文件中Kestra 内部通过容器名postgres_zoomcamp而不是host.docker.internal访问数据库pluginDefaults中相应修改。该 Compose 的要点包括三个 named volumepostgres-dataKestra 元数据、kestra-dataKestra 内部存储、zoomcamp-data练习数据kestra服务以user: root运行仅为访问 Docker socket 的开发态做法挂载docker.sock与/tmp/kestra-wdpostgres_zoomcamp提供练习库postgres-zoomcamp端口5432:5432pgadmin运行在 8085 端口避开 Kestra 的 8080/8081。完整 YAML 见原文档的 Docker Compose Example 折叠块。如果仍无法解决请停止并移除现有 Kestra Postgres 容器后重新docker-compose up -d。BigQuery CSV 列数不匹配错误如果遇到如下报错BigQueryError{reasoninvalid, locationnull, messageError 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官方给出了两条路径利用回填在 06_gcp_taxi_scheduled.yaml 上执行 Backfill时间范围设为2021-01-01至2021-07-31并分别对yellow与green各执行一次手动循环用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.ymldbt 工程示例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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考