# 1. 创建Python虚拟环境
$ python3 -m venv /bigdata/app/airflow/venv
$ source /bigdata/app/airflow/venv/bin/activate

# 2. 安装Airflow
$ pip install apache-airflow[celery,postgres,redis]==2.8.0 \
–constraint “https://raw.githubusercontent.com/apache/airflow/constraints-2.8.0/constraints-3.10.txt”

# 3. 配置环境变量
$ export AIRFLOW_HOME=/bigdata/app/airflow

# 4. 初始化数据库
$ airflow db init

# 5. 创建管理员用户
$ airflow users create \
–username admin \
–password admin123 \
–firstname Admin \
–lastname User \
–role Admin \
–email admin@fgedu.com

Admin user admin created

# 6. 配置airflow.cfg
$ cat > /bigdata/app/airflow/airflow.cfg << 'EOF' [core] dags_folder = /bigdata/app/airflow/dags base_log_folder = /bigdata/app/airflow/logs executor = CeleryExecutor [database] sql_alchemy_conn = postgresql+psycopg2://airflow:airflow123@fgedu-node1:5432/airflow_db [celery] broker_url = redis://fgedu-node1:6379/0 result_backend = db+postgresql://airflow:airflow123@fgedu-node1:5432/airflow_db [webserver] web_server_host = 0.0.0.0 web_server_port = 8080 workers = 4 [scheduler] scheduler_heartbeat_sec = 5 catchup_by_default = False EOF # 7. 创建PostgreSQL数据库 $ psql -U postgres postgres=# CREATE DATABASE airflow_db; postgres=# CREATE USER airflow WITH PASSWORD 'airflow123'; postgres=# GRANT ALL PRIVILEGES ON DATABASE airflow_db TO airflow; postgres=# \q # 8. 初始化数据库 $ airflow db init # 9. 启动Web Server $ airflow webserver --port 8080 & # 10. 启动Scheduler $ airflow scheduler & # 11. 启动Celery Worker $ airflow celery worker & # 12. 访问Web UI http://fgedu-node1:8080 用户名: admin 密码: admin123

# 1. 配置Redis集群
# 在fgedu-node1, fgedu-node2, fgedu-node3

$ cat > /etc/redis/redis.conf << 'EOF' bind 0.0.0.0 port 6379 cluster-enabled yes cluster-config-file nodes.conf cluster-node-timeout 5000 appendonly yes EOF $ systemctl restart redis # 2. 创建Redis集群 $ redis-cli --cluster create \ fgedu-node1:6379 fgedu-node2:6379 fgedu-node3:6379 \ --cluster-replicas 0 # 3. 配置PostgreSQL主从 # 主库配置 $ cat >> /var/lib/pgsql/data/postgresql.conf << 'EOF' wal_level = replica max_wal_senders = 3 EOF $ cat >> /var/lib/pgsql/data/pg_hba.conf << 'EOF' host replication airflow fgedu-node2/32 trust host replication airflow fgedu-node3/32 trust EOF $ systemctl restart postgresql # 从库配置 $ pg_basebackup -h fgedu-node1 -U airflow -D /var/lib/pgsql/data -P -R $ systemctl start postgresql # 4. 部署多Web Server # 在fgedu-node2 $ airflow webserver --port 8080 & # 5. 部署多Scheduler # 在fgedu-node2 $ airflow scheduler & # 6. 部署多Worker # 在fgedu-node2, fgedu-node3 $ airflow celery worker & # 7. 配置Nginx负载均衡 $ cat > /etc/nginx/conf.d/airflow.conf << 'EOF' upstream airflow_cluster { server fgedu-node1:8080; server fgedu-node2:8080; } server { listen 8080; server_name airflow.fgedu.net.cn; location / { proxy_pass http://airflow_cluster; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } } EOF $ systemctl restart nginx # 8. 访问高可用地址 http://airflow.fgedu.net.cn:8080

# 1. 创建DAG文件
$ cat > /bigdata/app/airflow/dags/fgedu_etl_dag.py << 'EOF' """ fgedu_etl_dag.py from:www.itpux.com.qq113257174.wx:itpux-com web: http://www.fgedu.net.cn ETL DAG示例 """ from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator from airflow.providers.apache.hive.operators.hive import HiveOperator from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator from airflow.utils.dates import days_ago from datetime import timedelta # 默认参数 default_args = { 'owner': 'fgedu', 'depends_on_past': False, 'email': ['admin@fgedu.com'], 'email_on_failure': True, 'email_on_retry': False, 'retries': 3, 'retry_delay': timedelta(minutes=5), } # 定义DAG dag = DAG( 'fgedu_etl_daily', default_args=default_args, description='Daily ETL DAG', schedule_interval='0 1 * * *', start_date=days_ago(1), catchup=False, tags=['etl', 'fgedu', 'daily'], ) # 数据采集任务 sqoop_import = BashOperator( task_id='sqoop_import', bash_command=''' sqoop import \ --connect jdbc:mysql://fgedu-node1:3306/fgedudb \ --username fgedu \ --password fgedu123 \ --table user_info \ --target-dir /user/fgedu/staging/user_info/{{ds}} \ --fields-terminated-by ',' \ --m 4 ''', dag=dag, ) # 数据清洗任务 hive_clean = HiveOperator( task_id='hive_clean', hive_cli_conn_id='hive_default', hql=''' INSERT OVERWRITE DIRECTORY '/user/fgedu/cleaned/user_info/{{ds}}' SELECT user_id, UPPER(name) as name, age, city FROM external_user_info WHERE dt = '{{ds}}' AND age > 0 AND age < 150 ''', dag=dag, ) # 数据转换任务 spark_transform = SparkSubmitOperator( task_id='spark_transform', application='/user/fgedu/lib/etl.jar', java_class='com.fgedu.etl.DataTransform', conn_id='spark_default', application_args=[ '/user/fgedu/cleaned/user_info/{{ds}}', '/user/fgedu/output/user_info/{{ds}}' ], conf={ 'spark.executor.memory': '4g', 'spark.executor.cores': '2', 'spark.num.executors': '10' }, dag=dag, ) # 数据验证任务 def validate_data(**context): import subprocess date = context['ds'] cmd = f'hdfs dfs -test -e /user/fgedu/output/user_info/{date}/_SUCCESS' result = subprocess.run(cmd, shell=True) if result.returncode != 0: raise Exception(f'Data validation failed for {date}') print(f'Data validation passed for {date}') validate = PythonOperator( task_id='validate', python_callable=validate_data, dag=dag, ) # 定义任务依赖 sqoop_import >> hive_clean >> spark_transform >> validate
EOF

# 2. 验证DAG
$ airflow dags list

dag_id | filepath
——————+————————
fgedu_etl_daily | fgedu_etl_dag.py

# 3. 测试DAG
$ airflow dags test fgedu_etl_daily 2026-04-08

[2026-04-08 15:00:00,000] {taskinstance.py:1234} INFO – Dependencies all met for
…
[2026-04-08 15:05:00,000] {taskinstance.py:1234} INFO – Task succeeded

# 4. 手动触发DAG
$ airflow dags trigger fgedu_etl_daily

Created

# 1. 创建分支DAG
$ cat > /bigdata/app/airflow/dags/fgedu_branch_dag.py << 'EOF' """ fgedu_branch_dag.py from:www.itpux.com.qq113257174.wx:itpux-com web: http://www.fgedu.net.cn 分支DAG示例 """ from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import BranchPythonOperator from airflow.utils.dates import days_ago def choose_branch(**context): import random branch = random.choice(['branch_a', 'branch_b']) print(f'Chosen branch: {branch}') return branch dag = DAG( 'fgedu_branch_dag', schedule_interval='0 2 * * *', start_date=days_ago(1), catchup=False, ) start = BashOperator( task_id='start', bash_command='echo "Start"', dag=dag, ) choose = BranchPythonOperator( task_id='choose_branch', python_callable=choose_branch, dag=dag, ) branch_a = BashOperator( task_id='branch_a', bash_command='echo "Branch A"', dag=dag, ) branch_b = BashOperator( task_id='branch_b', bash_command='echo "Branch B"', dag=dag, ) end = BashOperator( task_id='end', bash_command='echo "End"', trigger_rule='none_failed', dag=dag, ) start >> choose >> [branch_a, branch_b] >> end
EOF

# 2. 创建子DAG
$ cat > /bigdata/app/airflow/dags/fgedu_subdag_parent.py << 'EOF' """ fgedu_subdag_parent.py from:www.itpux.com.qq113257174.wx:itpux-com web: http://www.fgedu.net.cn 子DAG示例 """ from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.subdag import SubDagOperator from airflow.utils.dates import days_ago def subdag(parent_dag_name, child_dag_name, args): dag_subdag = DAG( dag_id=f'{parent_dag_name}.{child_dag_name}', default_args=args, schedule_interval=None, ) with dag_subdag: t1 = BashOperator( task_id='task1', bash_command='echo "SubDAG Task 1"', ) t2 = BashOperator( task_id='task2', bash_command='echo "SubDAG Task 2"', ) t1 >> t2

return dag_subdag

dag = DAG(
‘fgedu_subdag_parent’,
schedule_interval=’0 3 * * *’,
start_date=days_ago(1),
catchup=False,
)

start = BashOperator(
task_id=’start’,
bash_command=’echo “Start”‘,
dag=dag,
)

subdag_task = SubDagOperator(
task_id=’subdag_task’,
subdag=subdag(‘fgedu_subdag_parent’, ‘subdag_task’, dag.default_args),
dag=dag,
)

end = BashOperator(
task_id=’end’,
bash_command=’echo “End”‘,
dag=dag,
)

start >> subdag_task >> end
EOF

# 1. BashOperator
from airflow.operators.bash import BashOperator

bash_task = BashOperator(
task_id=’bash_task’,
bash_command=’echo “Hello Airflow”‘,
dag=dag,
)

# 2. PythonOperator
from airflow.operators.python import PythonOperator

def python_function(**context):
print(f”Execution date: {context[‘ds’]}”)
return ‘success’

python_task = PythonOperator(
task_id=’python_task’,
python_callable=python_function,
dag=dag,
)

# 3. HiveOperator
from airflow.providers.apache.hive.operators.hive import HiveOperator

hive_task = HiveOperator(
task_id=’hive_task’,
hive_cli_conn_id=’hive_default’,
hql=’SELECT COUNT(*) FROM user_info’,
dag=dag,
)

# 4. SparkSubmitOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

spark_task = SparkSubmitOperator(
task_id=’spark_task’,
application=’/user/fgedu/lib/etl.jar’,
java_class=’com.fgedu.etl.SparkETL’,
conn_id=’spark_default’,
dag=dag,
)

# 5. MySqlOperator
from airflow.providers.mysql.operators.mysql import MySqlOperator

mysql_task = MySqlOperator(
task_id=’mysql_task’,
mysql_conn_id=’mysql_default’,
sql=’SELECT COUNT(*) FROM user_info’,
dag=dag,
)

# 6. DockerOperator
from airflow.providers.docker.operators.docker import DockerOperator

docker_task = DockerOperator(
task_id=’docker_task’,
image=’python:3.10′,
command=’python -c “print(\’Hello Docker\’)”‘,
dag=dag,
)

# 7. Sensor
from airflow.sensors.filesystem import FileSensor

file_sensor = FileSensor(
task_id=’file_sensor’,
filepath=’/user/fgedu/input/data.csv’,
fs_conn_id=’hdfs_default’,
poke_interval=60,
timeout=3600,
dag=dag,
)

# 8. 自定义Operator
from airflow.models import BaseOperator

class FgeduCustomOperator(BaseOperator):
def __init__(self, name, **kwargs):
super().__init__(**kwargs)
self.name = name

def execute(self, context):
self.log.info(f’Executing custom operator: {self.name}’)
return f’Hello {self.name}’

custom_task = FgeduCustomOperator(
task_id=’custom_task’,
name=’Fgedu’,
dag=dag,
)