在数据处理场景中，经常需要先从远程地址下载XML文件，完成内容解析、数据清洗等处理后，再将结果上传到指定存储位置。使用Airflow DAG可以将这三个步骤串联成有序的工作流，实现流程的自动化调度和异常监控。
环境准备
首先需要确保Airflow环境已经安装完成，并且具备网络请求、XML解析、文件上传相关的依赖。如果是处理本地文件操作，不需要额外依赖；如果需要和远程服务交互，可能需要安装对应的Python库，比如requests用于下载文件，boto3用于上传到对象存储。
DAG整体结构设计
整个工作流分为三个核心任务，任务之间的依赖关系为：下载任务执行成功后，才能执行处理任务；处理任务执行成功后，才能执行上传任务。我们可以通过PythonOperator来定义每个任务的具体执行逻辑。
任务依赖关系
- download_xml：下载远程XML文件到本地临时目录
- process_xml：读取本地XML文件，完成解析和处理，生成结果文件
- upload_result：将处理后的结果文件上传到目标存储位置
完整DAG代码实现
下面是完整的DAG代码示例，包含三个任务的定义和依赖配置：
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import requests
import xml.etree.ElementTree as ET
import os
# 定义默认参数
default_args = {
'owner': 'airflow',
'start_date': datetime(2024, 1, 1),
'retries': 1,
}
# 下载XML文件的任务函数
def download_xml_func(**kwargs):
# 远程XML文件地址，实际使用时替换为真实地址
xml_url = "http://ipipp.com/data/test.xml"
# 本地临时存储路径
local_path = "/tmp/raw_data.xml"
response = requests.get(xml_url)
response.raise_for_status()
with open(local_path, 'w', encoding='utf-8') as f:
f.write(response.text)
# 将本地路径传递到下游任务
kwargs['ti'].xcom_push(key='xml_local_path', value=local_path)
# 处理XML文件的任务函数
def process_xml_func(**kwargs):
ti = kwargs['ti']
# 获取上游任务传递的XML文件路径
xml_path = ti.xcom_pull(key='xml_local_path', task_ids='download_xml')
# 解析XML文件
tree = ET.parse(xml_path)
root = tree.getroot()
# 示例处理逻辑：提取所有标签为item的内容，拼接成字符串
result_content = ""
for item in root.findall('item'):
result_content += item.text + "n"
# 保存处理结果到本地
result_path = "/tmp/processed_result.txt"
with open(result_path, 'w', encoding='utf-8') as f:
f.write(result_content)
# 将结果文件路径传递到下游任务
ti.xcom_push(key='result_local_path', value=result_path)
# 上传结果文件的任务函数
def upload_result_func(**kwargs):
ti = kwargs['ti']
# 获取上游任务传递的结果文件路径
result_path = ti.xcom_pull(key='result_local_path', task_ids='process_xml')
# 示例上传逻辑：这里模拟上传到本地另一个目录，实际可替换为对象存储上传逻辑
target_dir = "/tmp/uploaded/"
if not os.path.exists(target_dir):
os.makedirs(target_dir)
target_path = os.path.join(target_dir, os.path.basename(result_path))
with open(result_path, 'r', encoding='utf-8') as f_src:
with open(target_path, 'w', encoding='utf-8') as f_dst:
f_dst.write(f_src.read())
print(f"文件已上传到{target_path}")
# 初始化DAG
with DAG(
dag_id='xml_process_dag',
default_args=default_args,
schedule_interval=None,
catchup=False,
tags=['xml', 'file_process'],
) as dag:
# 定义下载任务
download_task = PythonOperator(
task_id='download_xml',
python_callable=download_xml_func,
provide_context=True,
)
# 定义处理任务
process_task = PythonOperator(
task_id='process_xml',
python_callable=process_xml_func,
provide_context=True,
)
# 定义上传任务
upload_task = PythonOperator(
task_id='upload_result',
python_callable=upload_result_func,
provide_context=True,
)
# 设置任务依赖
download_task >> process_task >> upload_task
代码说明
上述代码中，我们使用xcom_push和xcom_pull方法在任务之间传递文件路径参数，避免硬编码路径导致的问题。每个任务函数都做了独立的逻辑封装，方便后续单独修改某个步骤的实现。
如果XML文件下载需要认证，可以在download_xml_func中添加请求头或者认证参数；如果处理结果需要上传到云存储，只需要修改upload_result_func中的上传逻辑，替换为对应SDK的上传方法即可。
运行验证
将DAG文件放到Airflow的dags目录下，等待Airflow加载完成后，在Web界面中触发DAG运行。可以查看每个任务的执行日志，确认文件下载、处理、上传三个步骤是否按顺序执行成功。如果某个任务失败，Airflow会自动重试，也可以手动触发重新执行失败的任务。