1. 项目概述这不是又一本Python教程而是一份2025年真实生产环境的数据流水线作战手册“Mastering Python Data Pipelines in 2025”这个标题里“Mastering”不是指学会几个pandas函数而是指你能在凌晨三点收到告警邮件时不翻文档、不查Stack Overflow直接SSH进服务器三分钟定位到是Airflow DAG的某个task因上游API返回了非标准JSON结构而卡死顺手写个轻量级schema校验wrapper加进DAG依赖链重启后流水线自动恢复——整个过程像拧紧一颗松动的螺丝一样自然。我带过七支数据工程团队从金融风控实时反欺诈系统到东南亚电商千万级SKU的动态定价引擎再到医疗影像AI标注平台的元数据治理流水线所有这些系统在2025年的共同特征是Python不再是胶水语言而是整条数据流水线的承重骨架。它既要处理每秒上万条IoT设备上报的时序数据流也要调度跨云AWS 阿里云私有K8s的混合计算任务还要在满足GDPR和国内《个人信息保护法》双重合规要求下对敏感字段做动态脱敏与血缘追踪。这意味着2025年的Python数据流水线必须同时扛住三重压力性能不能靠堆机器硬扛架构不能靠改配置临时缝合合规不能靠人工审计事后补救。如果你还在用for row in df.iterrows()遍历百万行数据或者把Airflow DAG写成一个300行的巨型Python脚本或者认为“数据质量跑完ETL没报错”那这份手册就是为你准备的。它不讲Python基础语法不罗列所有开源库只聚焦2025年一线团队每天真正在用、正在调、正在骂、也正在依赖的核心技术组合与实战逻辑——从本地开发环境的一键复现到生产集群的灰度发布再到故障时的秒级回滚。适合两类人一类是已能熟练写pandas和SQL但一碰Airflow就卡在Docker Compose网络配置上的中级工程师另一类是技术负责人需要快速判断团队当前的流水线架构是否已在2025年的技术水位线下沉——比如你的DAG是否还依赖BashOperator去调用一个shell脚本启动Spark作业如果是那它大概率已经成了你系统里最不可控的单点故障源。2. 核心设计思路为什么2025年必须放弃“脚本式流水线”转向“声明式可验证”范式2.1 从“能跑通”到“可验证”的范式迁移2024年之前很多团队的数据流水线设计哲学是“先让它跑起来”。典型做法是用Python脚本读取CSVpandas清洗SQL写入MySQL最后用cron定时执行。这种模式在小规模、低频次、业务逻辑简单的场景下确实高效。但到了2025年当你的数据源从5个暴增至37个含SaaS API、IoT MQTT Broker、CDC捕获的PostgreSQL WAL日志、第三方CDN日志、内部微服务gRPC流当单日数据量从GB级跃升至TB级当业务方要求“任意时间点数据快照可追溯、任意字段变更可归因”旧范式立刻崩塌。我亲眼见过一家跨境电商公司其核心订单流水线因一个未被发现的fillna(0)操作将数万条实际为NULL的优惠券ID强制转为0导致下游推荐系统持续一周向用户推送错误优惠损失预估超两百万。问题根源不在代码本身而在于整个流水线缺乏可验证性——没有schema定义没有数据质量断言没有版本化元数据。因此2025年设计的第一原则是一切流水线组件必须自带“自证清白”的能力。这直接决定了我们放弃纯脚本转向以DAG为核心、以YAML/JSON Schema为契约、以Pydantic V2为校验引擎的声明式架构。例如一个典型的2025年数据源定义不再是一行pd.read_csv(s3://bucket/orders.csv)而是一个严格约束的Pydantic模型from pydantic import BaseModel, Field, validator from datetime import datetime from typing import Optional, List class OrderSourceSchema(BaseModel): order_id: str Field(..., min_length12, max_length32, description全局唯一订单号格式ORD-YYYYMMDD-XXXXXX) user_id: int Field(..., ge100000, le999999999, description用户ID6-9位正整数) total_amount: float Field(..., gt0.0, lt1000000.0, description订单总金额单位元精度2位小数) created_at: datetime Field(..., description订单创建时间ISO 8601格式) coupon_code: Optional[str] Field(None, regexr^[A-Z]{3}-\d{4}$, description优惠券码格式AAA-1234可为空) validator(total_amount) def round_to_two_decimals(cls, v): return round(v, 2) class Config: extra forbid # 严格禁止未知字段这是数据契约的底线这个模型不是文档而是运行时强制校验器。当上游API返回{order_id:ORD-20250401-000001, user_id:123456, total_amount:99.999, created_at:2025-04-01T10:30:00Z, extra_field:junk}时流水线会在OrderSourceSchema.parse_obj()这一步直接抛出ValidationError并精确指出extra_field非法、total_amount精度超限。这比任何日志告警都早三步——它发生在数据进入处理逻辑之前而非写入数据库之后。这就是“可验证”的力量它把数据质量控制点前移到了入口而不是靠下游报表异常去倒推。2.2 工具链选型为什么是Airflow 2.9 Prefect 3.0 Dagster 1.7的混合部署2025年没有“银弹”调度器。我见过太多团队在选型时陷入非此即彼的误区要么All-in Airflow结果DAG越来越臃肿调试一次要等五分钟要么All-in Prefect结果在复杂依赖图如“只有当A、B、C三个上游任务全部成功且D任务失败次数2次时才触发E任务”面前束手无策。真实生产环境需要的是分层混用。我们的标准配置是Airflow 2.9 作为“主干调度器”负责跨系统、长周期、高可靠性的核心编排。比如每日02:00触发全量数据同步从Oracle到Snowflake每周日凌晨触发模型重训练流水线调用SageMaker API每月1日生成财务对账报告连接ERP和BI系统。选择Airflow是因为它的Web UI成熟度、RBAC权限体系、以及与Kubernetes Executor的深度集成在2025年仍是企业级调度的事实标准。关键升级点是启用TriggerRule的all_done和none_failed_or_skipped组合并配合TaskGroup重构DAG将过去一个300行的etl_dag.py拆解为extract_group,transform_group,load_group三个逻辑组每个组内任务可独立启停、重试、查看日志。Prefect 3.0 作为“弹性工作流引擎”专攻那些Airflow难以优雅处理的场景。典型如实时事件驱动的短时任务用户下单后10秒内完成库存预占与风控初筛、需要复杂条件分支的动态流程根据订单金额自动选择不同的支付清算通道、或需嵌入Python原生异步逻辑的场景并发调用10个不同银行的API查询余额。Prefect的flow和task装饰器让逻辑极其清晰其内置的State机制如Running,Completed,Failed,Paused比Airflow的upstream_failed更细粒度。更重要的是Prefect 3.0的Deployment概念完美解决了Airflow中“DAG代码更新后如何灰度”的痛点——你可以为同一段flow代码创建多个Deployment分别指向staging和production基础设施通过UI一键切换流量比例。Dagster 1.7 作为“数据资产编排层”当你的核心诉求从“任务调度”升级为“数据资产治理”时Dagster不可替代。它强制你以Asset资产为中心建模而非以Task任务为中心。例如一个orders_fact_table资产其定义不仅包含如何构建它asset函数还必须声明其上游依赖orders_raw_json,users_dim_table、下游消费者revenue_dashboard,fraud_model_training_dataset、以及数据质量检查asset_check。Dagster的AssetGraph会自动生成完整的数据血缘图点击任意节点即可看到该资产的最近一次更新时间、耗时、输入输出记录、甚至原始SQL。这直接支撑了2025年最严苛的合规要求当法务部问“用户手机号字段在哪些地方被使用过”你能在10秒内给出完整清单而非花三天翻代码库。提示混合部署的关键不是“多装几个工具”而是明确边界。我们的红线是任何需要写入生产数据库、影响核心业务指标、或涉及资金结算的任务必须由Airflow调度任何需要毫秒级响应、高度动态分支、或强依赖Python生态如scikit-learn模型的任务交给Prefect任何需要回答“这个数据从哪来、到哪去、谁在用、质量如何”问题的资产必须注册为Dagster Asset。三者通过REST API和消息队列我们用Apache Pulsar松耦合通信避免形成新的单点故障。2.3 架构分层为什么“计算与存储分离”已成铁律而“代码与配置分离”是生存底线2025年再把Spark作业打包进Docker镜像随Airflow Worker一起部署是自杀行为。原因很简单计算资源需求与代码迭代频率完全错配。一个Spark作业可能需要32核128G内存但它的Python逻辑可能一个月只改一次而Airflow Worker只需2核4G却要每天更新DAG定义。强行捆绑导致每次DAG小修都要重建大镜像、推送Registry、滚动更新Worker Pod——一次发布耗时15分钟期间所有新DAG都无法调度。因此2025年架构的铁律是计算与存储必须物理分离代码与配置必须逻辑分离。计算分离实践我们将所有ETL/ML任务抽象为标准化的“计算单元”Compute Unit。每个单元是一个轻量级Python包如etl_orders_v2仅包含业务逻辑和依赖声明requirements.txt不包含任何调度或基础设施代码。它被构建为OCI镜像推送到私有Registry。Airflow的KubernetesPodOperator在运行时按需拉取该镜像启动一个独立Pod执行任务完成后立即销毁。这样etl_orders_v2的镜像可以被Airflow、Prefect、甚至CI/CD Pipeline复用。我们甚至用它来支持本地开发make run-etl-orders命令会启动一个本地K8s Pod挂载当前代码目录实现“所写即所跑”。配置分离实践所有环境相关参数数据库URL、API密钥、超时阈值、重试次数绝不硬编码在Python文件中。我们采用三层配置体系基础层Base存于Git仓库的config/base.yaml定义所有环境共有的默认值如default_timeout: 300,max_retries: 3。环境层Env存于Git仓库的config/staging.yaml和config/production.yaml覆盖基础层如staging中db_url: postgresql://test:testpg-staging:5432/dbproduction中db_url: postgresql://prod:xxxpg-prod:5432/db。密钥层Secret绝不存于Git由HashiCorp Vault统一管理Airflow/Prefect/Dagster通过Vault Agent Sidecar自动注入环境变量。例如AIRFLOW_CONN_AWS_DEFAULT这个连接字符串由Vault动态生成并注入Python代码只通过os.getenv(AIRFLOW_CONN_AWS_DEFAULT)读取。这套体系带来的好处是git diff能看到所有配置变更git blame能精准定位谁在何时改了哪个环境的哪个参数而密钥永远不落地、不泄露、可轮换。我曾帮一家客户将配置错误导致的生产事故从月均3次降至零——因为他们终于能看清上次“数据延迟”根本不是代码bug而是production.yaml里一个batch_size: 1000被误写成了batch_size: 100导致Kafka消费者吞吐骤降。3. 核心实操环节从零搭建一条符合2025标准的端到端订单流水线3.1 环境准备用Docker Compose一键拉起本地开发沙盒2025年的本地开发必须无限逼近生产环境。我们拒绝“本地用SQLite生产用PostgreSQL”这种自欺欺人的方案。目标是在MacBook上用一条docker compose up -d命令启动包含Airflow Webserver、Scheduler、PostgreSQL元数据库、Redis Celery Broker、MinIO对象存储、Pulsar消息队列、以及一个预装了所有依赖的JupyterLab的完整沙盒。所有组件版本严格对齐生产集群Airflow 2.9.2, PostgreSQL 15, MinIO RELEASE.2024-09-12T00-00-00Z。docker-compose.yml的核心片段如下version: 3.8 services: # Airflow核心组件 airflow-webserver: image: apache/airflow:2.9.2 environment: - AIRFLOW__CORE__EXECUTORKubernetesExecutor - AIRFLOW__CORE__SQL_ALCHEMY_CONNpostgresqlpsycopg2://airflow:airflowpostgres/airflow - AIRFLOW__CELERY__RESULT_BACKENDredis://redis:6379/0 - AIRFLOW__KUBERNETES__NAMESPACEdefault # ... 其他20个必要环境变量 volumes: - ./dags:/opt/airflow/dags # 挂载DAG代码 - ./plugins:/opt/airflow/plugins depends_on: - postgres - redis # 生产级对象存储替代S3 minio: image: quay.io/minio/minio:RELEASE.2024-09-12T00-00-00Z command: server /data --console-address :9001 environment: - MINIO_ROOT_USERminioadmin - MINIO_ROOT_PASSWORDminioadmin ports: - 9000:9000 - 9001:9001 volumes: - minio_data:/data # 消息队列用于Airflow-Prefect-Dagster通信 pulsar: image: apachepulsar/pulsar:3.2.0 command: bin/pulsar standalone ports: - 6650:6650 # Broker - 8080:8080 # Admin API volumes: - pulsar_data:/pulsar/data volumes: minio_data: pulsar_data: postgres_data:启动后访问http://localhost:8080Airflow UI、http://localhost:9001MinIO Console、http://localhost:8080Pulsar Admin即可。关键技巧是所有DAG代码都放在./dags目录下Airflow会自动热加载。当你修改dags/etl_orders_dag.py并保存Airflow Scheduler几秒内就能检测到变更无需重启容器。这极大提升了迭代效率。我建议新手先不要碰Kubernetes Executor而是将AIRFLOW__CORE__EXECUTOR临时改为LocalExecutor用airflow tasks test etl_orders_dag extract_task 2025-04-01命令在本地测试单个task确认逻辑无误后再切回Kubernetes模式。3.2 数据源接入用Pydantic Schema Great Expectations实现“带契约的摄入”假设我们要接入一个第三方电商平台的订单API其文档声称返回JSON格式但实际经常返回非标准数据如total_amount有时是字符串99.99有时是数字99.99有时缺失。2025年的正确做法不是写一堆if isinstance(data[total_amount], str): ...而是用契约先行。步骤如下定义Pydantic Schema如前文OrderSourceSchema并将其保存为schemas/order_source.py。编写摄入任务Ingest Task这是一个独立的Python模块不依赖Airflow可单独测试。# ingest_orders.py import requests import json from datetime import datetime, timedelta from schemas.order_source import OrderSourceSchema from typing import List, Dict, Any def fetch_orders_from_api(start_date: str, end_date: str) - List[Dict[str, Any]]: 从API获取指定日期范围的订单返回原始JSON列表 url https://api.example-shop.com/v1/orders params {start_date: start_date, end_date: end_date} headers {Authorization: fBearer {get_api_token()}} response requests.get(url, paramsparams, headersheaders, timeout30) response.raise_for_status() return response.json() def validate_and_normalize_orders(raw_orders: List[Dict]) - List[OrderSourceSchema]: 用Pydantic Schema校验并标准化原始数据 validated_orders [] for i, raw in enumerate(raw_orders): try: # 强制转换为标准格式 if isinstance(raw.get(total_amount), str): raw[total_amount] float(raw[total_amount]) # 解析时间字符串 if isinstance(raw.get(created_at), str): raw[created_at] datetime.fromisoformat(raw[created_at].replace(Z, 00:00)) # 创建实例触发所有校验 order OrderSourceSchema.parse_obj(raw) validated_orders.append(order) except Exception as e: # 记录详细错误便于后续分析 print(fValidation failed for order #{i}: {e}, raw data: {raw}) # 可选择跳过或存入quarantine bucket continue return validated_orders def save_to_minio(validated_orders: List[OrderSourceSchema], bucket: str, key: str): 将校验后的数据保存为Parquet到MinIO import pandas as pd from minio import Minio # 转为DataFrame利用PyArrow自动处理datetime和nullable类型 df pd.DataFrame([o.dict() for o in validated_orders]) # 写入Parquet分区为year/month/day partition_cols [created_at] # 使用minio-py SDK上传 client Minio(localhost:9000, minioadmin, minioadmin, secureFalse) # ... Parquet写入逻辑集成Great Expectations进行数据质量断言在validate_and_normalize_orders之后加入GE检查from great_expectations.core.batch import RuntimeBatchRequest from great_expectations.data_context import BaseDataContext from great_expectations.data_context.types.base import ( DataContextConfig, FilesystemStoreBackendDefaults, ) def run_data_quality_checks(df: pd.DataFrame) - Dict: 运行预定义的数据质量检查集 # 初始化GE上下文指向本地配置 context BaseDataContext( project_configDataContextConfig( store_backend_defaultsFilesystemStoreBackendDefaults(root_directory./great_expectations/) ) ) # 创建RuntimeBatchRequest batch_request RuntimeBatchRequest( datasource_namemy_minio_datasource, data_connector_namedefault_inferred_data_connector_name, data_asset_nameorders_parquet, # 这个名称需与GE配置匹配 runtime_parameters{path: s3a://my-bucket/orders/2025/04/01/part-00000.parquet}, batch_identifiers{default_identifier_name: ingest_20250401}, ) # 执行检查 validator context.get_validator(batch_requestbatch_request) results validator.expect_column_values_to_not_be_null(columnorder_id) results validator.expect_column_values_to_be_between(columntotal_amount, min_value0.01, max_value100000.0) results validator.expect_column_proportion_of_unique_values_to_be_greater_than( columnorder_id, min_value0.999 ) return results.to_json_dict()这个流程确保数据在进入流水线第一站时就完成了格式标准化、业务规则校验、以及统计质量评估。任何不符合契约的数据都会被拦截、记录、并通知数据工程师而不是污染下游。3.3 流水线编排Airflow DAG Prefect Flow Dagster Asset的协同工作流现在我们将上述摄入任务整合进一个端到端流水线。它分为三个阶段由三个工具协同完成Stage 1: Airflow调度摄入Airflow DAG# dags/etl_orders_dag.py from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator from datetime import datetime, timedelta default_args { owner: data-engineering, depends_on_past: False, start_date: datetime(2025, 1, 1), email_on_failure: True, retries: 1, retry_delay: timedelta(minutes5), } dag DAG( etl_orders_pipeline, default_argsdefault_args, descriptionEnd-to-end orders ETL pipeline, schedule_interval0 2 * * *, # 每天02:00 catchupFalse, tags[etl, orders], ) # Task 1: 在K8s Pod中运行摄入脚本 ingest_task KubernetesPodOperator( task_idingest_orders_from_api, nameingest-orders, namespacedefault, imagemy-registry/ingest-orders:v1.2.0, # 预构建的镜像 cmds[python, ingest_orders.py], arguments[--start-date, {{ ds }}, --end-date, {{ ds }}], get_logsTrue, is_delete_operator_podTrue, in_clusterFalse, # 本地开发用KubeConfig config_file/Users/me/.kube/config, do_xcom_pushFalse, dagdag, )Stage 2: Prefect处理动态逻辑Prefect Flow# flows/process_orders_flow.py from prefect import flow, task from prefect.tasks import task_input_hash from datetime import datetime task(cache_key_fntask_input_hash, cache_expirationtimedelta(hours1)) def load_orders_from_minio(date_str: str) - pd.DataFrame: 从MinIO加载当日订单Parquet # ... 实现 task def enrich_orders_with_user_data(df: pd.DataFrame) - pd.DataFrame: 调用内部用户服务API补充用户等级、地域信息 # ... 实现 flow(nameprocess-orders-flow) def process_orders_flow(date_str: str): df load_orders_from_minio(date_str) enriched_df enrich_orders_with_user_data(df) # 将结果写入Snowflake write_to_snowflake(enriched_df, tableorders_enriched) # 在Airflow中触发Prefect Flow # dags/etl_orders_dag.py 中添加 from prefect.client import get_client from prefect.deployments import Deployment async def trigger_prefect_flow(**context): async with get_client() as client: deployment await client.read_deployment_by_name(process-orders-flow/staging) await client.create_flow_run_from_deployment( deployment.id, parameters{date_str: context[ds]}, ) trigger_prefect_task PythonOperator( task_idtrigger_prefect_process_flow, python_callabletrigger_prefect_flow, dagdag, ) ingest_task trigger_prefect_taskStage 3: Dagster注册为资产Dagster Asset# assets/orders_assets.py from dagster import asset, AssetIn, AssetOut, multi_asset from dagster_pyspark import PySparkResource from pyspark.sql import DataFrame asset( ins{orders_enriched: AssetIn(key_prefix[snowflake, raw])}, outs{orders_fact: AssetOut( io_manager_keysnowflake_io_manager, key_prefix[snowflake, mart], description事实表聚合后的订单事实按天分区 )} ) def orders_fact(orders_enriched: DataFrame) - DataFrame: 构建订单事实表 return (orders_enriched .groupBy(date_trunc(day, created_at)) .agg( F.sum(total_amount).alias(daily_revenue), F.count(*).alias(order_count), F.avg(total_amount).alias(avg_order_value) )) # 定义数据质量检查 asset_check(assetorders_fact) def check_orders_fact_not_empty(orders_fact: DataFrame): count orders_fact.count() assert count 0, fOrders fact table is empty for date {orders_fact.first().date_trunc} return AssetCheckResult(passedcount 0, metadata{row_count: count})这个协同工作流的价值在于每个工具各司其职且彼此解耦。Airflow保证摄入的可靠调度Prefect提供灵活的动态处理Dagster则为最终产出的orders_fact表建立完整的资产档案。当业务方质疑“昨天的GMV为什么比前天低20%”你可以在Dagster UI中点击orders_fact立刻看到它依赖的orders_enriched表在昨天的更新时间、耗时、以及所有相关的数据质量检查结果包括check_orders_fact_not_empty是否通过排查效率提升十倍。3.4 生产部署与监控用Prometheus Grafana OpenTelemetry构建可观测性闭环2025年没有监控的流水线等于没有刹车的汽车。我们的监控体系覆盖三层基础设施层Infra用Prometheus抓取Airflow Scheduler/Worker、PostgreSQL、MinIO、Pulsar的指标CPU、内存、磁盘IO、网络延迟、队列长度。例如airflow_scheduler_dagbag_import_duration_seconds指标能告诉你DAG解析是否变慢minio_bucket_objects_total能监控对象数量突增。应用层App在Airflow DAG中注入OpenTelemetry Tracing。每个task开始时创建一个Span记录其输入参数、执行耗时、状态success/fail/skip、以及下游依赖。代码示例from opentelemetry import trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor # 初始化Tracer在DAG文件顶部 provider TracerProvider() processor BatchSpanProcessor(OTLPSpanExporter(endpointhttp://otel-collector:4318/v1/traces)) provider.add_span_processor(processor) trace.set_tracer_provider(provider) tracer trace.get_tracer(__name__) # 在task中使用 def my_etl_task(**context): with tracer.start_as_current_span(etl_orders_transform) as span: span.set_attribute(dag_id, context[dag].dag_id) span.set_attribute(task_id, context[task].task_id) span.set_attribute(execution_date, context[ds]) # ... 执行业务逻辑 span.set_attribute(output_rows, len(result_df))业务层Biz用Grafana Dashboard展示核心业务指标。我们有一个名为“Data Pipeline Health”的看板包含SLA Compliance Rate按DAG计算按时完成的DAG数 / 总DAG数* 100%。阈值95%。Data Freshness每个关键资产如orders_fact距离最新分区的时间差。阈值15分钟。Quality Pass Rate所有Dagster Asset Check的通过率。阈值100%。Error Root Cause Distribution饼图显示错误来源API超时、Schema校验失败、数据库连接池满、内存溢出等。当SLA Compliance Rate跌破90%Grafana自动触发Alert发送到企业微信并创建Jira Ticket分配给值班工程师。这个Ticket会自动附带告警时间、受影响的DAG、最近3次执行的日志链接、以及OpenTelemetry Trace ID可直接在Jaeger中查看完整调用链。我见过最惊险的一次故障Trace显示99%的ingest_orders_from_api任务都在requests.get()上卡住但curl -v测试API又正常。最终通过Trace的http.status_code标签发现API返回了HTTP 429Too Many Requests而我们的代码没处理这个状态码导致requests库一直重试直到超时。修复方案很简单在fetch_orders_from_api中增加response.status_code 429的判断并加入指数退避。整个过程从告警到修复上线不到20分钟。4. 常见问题与实战排障一线工程师踩过的坑与独家技巧4.1 “DAG在本地能跑生产环境就失败”——环境一致性陷阱这是2025年最高频的问题。表面看是“环境不同”深层原因是Python依赖版本漂移。例如本地用pandas2.2.0生产用pandas2.1.4而后者不支持pd.concat(..., ignore_indexTrue)的新参数。我们的解决方案是所有Python环境必须基于pyproject.toml和poetry.lock文件构建且lock文件必须提交到Git。Poetry配置示例pyproject.toml[tool.poetry] name etl-orders version 1.2.0 description Orders ETL pipeline [tool.poetry.dependencies] python ^3.11 pandas { version ^2.2.0, allow-prereleases false } requests ^2.31.0 pydantic { version ^2.6.0, allow-prereleases false } great-expectations ^0.18.0 [tool.poetry.group.dev.dependencies] pytest ^7.4.0 black ^24.1.0 [build-system] requires [poetry-core] build-backend poetry.core.masonry.api构建Docker镜像时强制使用lock文件FROM python:3.11-slim WORKDIR /app COPY poetry.lock pyproject.toml ./ RUN pip install poetry \ poetry install --no-dev --without dev,test # 只安装生产依赖 COPY . . CMD [python, ingest_orders.py]实操心得我们要求所有DAG代码的requirements.txt必须由poetry export -f requirements.txt --without-hashes requirements.txt生成并提交。CI/CD Pipeline在构建镜像前会运行pip check如果发现依赖冲突如pandas和numpy版本不兼容立即失败。这杜绝了“本地好好的上线就炸”的魔咒。我曾帮一个团队将此类问题从月均5次降至零他们之前的做法是手动维护requirements.txt版本号全靠记忆。4.2 “Airflow UI卡死Scheduler日志刷屏”——元数据库性能瓶颈当Airflow Scheduler日志疯狂打印INFO - Processing file ...且Web UI打开一个DAG页面要等半分钟基本可以确定是PostgreSQL元数据库撑不住了。根本原因有两个一是dag_run和task_instance表数据爆炸每天数万条记录二是serialized_dag表未优化。我们的修复四步法清理历史DAG Run创建一个Airflow DAG每天凌晨执行DELETE FROM dag_run WHERE execution_date NOW() - INTERVAL 30 days; DELETE FROM task_instance WHERE execution_date NOW() - INTERVAL 30 days;注意DELETE比TRUNCATE安全因为它尊重外键约束。优化serialized_dag表该表存储DAG的序列化Python代码是巨大的BLOB。我们禁用store_serialized_dagsAIRFLOW__CORE__STORE_SERIALIZED_DAGSFalse改用AIRFLOW__CORE__STORE_DAG_CODETrue让Airflow直接从Git仓库读取DAG文件。这节省了90%的元数据库空间。为关键索引添加在task_instance表上为(dag_id, state, execution_date)创建复合索引在dag_run表上为(state,