Airflow的DAG为什么用Python写——Python元编程基础

Airflow DAG的本质——Python脚本

# Airflow的DAG文件就是一个Python脚本
# 调度器import它时，Python解释器执行所有代码
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
 
# 用Python变量定义配置
tables = ['events', 'users', 'devices', 'transactions']
 
# 用Python代码动态创建DAG
with DAG('daily_etl', start_date=datetime(2024,1,1),
         schedule='@daily') as dag:
 
    # 用for循环批量创建Task！
    for table in tables:
        PythonOperator(
            task_id=f'process_{table}',
            python_callable=process_table,
            op_kwargs={'table': table}
        )
 
# 4个Task用4行代码搞定
# 如果用YAML/XML配置，需要写4段几乎重复的配置
# 如果有100个表，Python循环100次，配置文件要写100段

Python元编程核心概念

① 一切皆对象：Python中class、function、module都是对象。DAG()返回的是一个对象，PythonOperator()创建的也是对象。你可以把它们存在列表里、传给函数、动态创建。

② type()——类的类：Python中class是由type创建的。type('MyClass', (Base,), {'x': 1})在运行时动态创建一个类。Airflow内部用类似的机制动态注册Operator。

③ __dict__——对象的属性字典：每个Python对象有一个__dict__存储它的所有属性。你可以在运行时给对象动态添加属性。Airflow的DAG对象在创建后可以动态添加Task。

# type()动态创建类
MyTask = type('MyTask', (PythonOperator,), {
    'default_args': {'retries': 3},
})
 
# __dict__动态添加属性
dag = DAG('my_dag')
dag.custom_config = {'priority': 'high'}  # 运行时添加属性
 
# __getattr__——属性访问的钩子
class Config:
    def __getattr__(self, name):
        return f'default_value_for_{name}'
c = Config()
print(c.any_key)  # 'default_value_for_any_key'

Airflow 2.0+的TaskFlow API——装饰器元编程

Airflow 2.0引入了TaskFlow API，用@task装饰器把普通Python函数变成Airflow Task：

from airflow.decorators import dag, task
from datetime import datetime
 
@dag(schedule='@daily', start_date=datetime(2024,1,1))
def daily_etl():
 
    @task
    def extract(table: str):
        return read_from_db(table)
 
    @task
    def transform(data):
        return clean_and_transform(data)
 
    @task
    def load(data):
        write_to_warehouse(data)
 
    # 用Python表达式定义依赖关系
    for table in ['events', 'users', 'devices']:
        raw = extract(table)
        cleaned = transform(raw)
        load(cleaned)
 
daily_etl()  # 调用函数创建DAG
 
# @task装饰器的魔法：
# 把普通函数包装成Airflow的PythonOperator
# 函数的返回值自动通过XCom传递给下一个Task
# 函数的参数自动从上游Task的XCom获取
# 这就是Python装饰器+元编程的威力

动态DAG生成——元编程的实战应用

在某短视频平台用Airflow时，我用Python动态生成了100+个ETL DAG：

# 从配置文件读取所有表的ETL配置
import yaml
with open('etl_config.yaml') as f:
    configs = yaml.safe_load(f)
 
# 动态生成每张表的DAG
for config in configs:
    dag_id = f'etl_{config["table"]}'
    schedule = config.get('schedule', '@daily')
    
    # exec()或globals()动态注册DAG
    # Airflow要求DAG对象在模块的全局作用域
    globals()[dag_id] = create_etl_dag(
        dag_id=dag_id,
        table=config['table'],
        schedule=schedule,
        dependencies=config.get('dependencies', [])
    )
 
# 100+张表只需要维护一个yaml配置文件
# Python脚本自动生成100+个DAG
# 新增一张表只需要在yaml里加一行

效果：维护100+个ETL任务的工作量从'每个任务一个DAG文件'变成'一个yaml配置+一个生成脚本'。新增表的上线时间从半天降到5分钟。

面试高频追问

追问一：Airflow为什么不用YAML定义DAG？

因为YAML是静态的——不能有循环、条件判断、函数调用。Python代码可以动态生成任意复杂的DAG结构。比如'根据上游表的分区数动态创建对应数量的Task'——这在YAML里做不到。不过DolphinScheduler和DataWorks选择了可视化+配置的方式，适合不太会写代码的用户。

追问二：metaclass是什么？Airflow用了吗？

metaclass是'类的类'——控制类的创建过程。class Foo(metaclass=Meta)时Python用Meta.__new__()创建Foo类。Airflow的DAG类用了一些元编程技巧（比如自动注册到全局DAG列表），但没有显式使用metaclass。Django的Model和SQLAlchemy的ORM是metaclass的经典应用。大数据开发不需要深入metaclass。

追问三：__init__和__new__有什么区别？

__new__创建对象实例（分配内存），__init__初始化对象（设置属性）。大部分情况只需要__init__。__new__的典型用途是单例模式和不可变类型的创建。面试问到说清楚区别就行，大数据开发极少需要自己写__new__。

总结

元编程面试讲三层：① Python一切皆对象（class/function/module都是对象，可以动态创建和修改）；② Airflow DAG的设计原理（Python脚本import时动态创建DAG/Task对象，比YAML灵活得多）；③ 实战应用（for循环批量生成Task、从配置文件动态生成100+ DAG、@task装饰器把函数变Task）。能讲出动态生成100+个DAG的实际经验，面试官会觉得你善于用Python特性解决工程问题。

下一篇聊Python面向对象：数据质量校验脚本写了500行怎么重构？

你们Airflow DAG怎么组织的？有动态生成DAG吗？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，只讲大数据面试干货