Airflow实战：如何将XML文件加载至BigQuery？

一、先修正你的XML转JSON代码 #

你的现有代码存在缩进错误、手动关闭文件（上下文管理器with会自动处理文件关闭），而且在Airflow场景下没必要生成本地JSON文件——分布式环境中本地文件不可靠，直接在内存处理数据更合理。修正并优化后的代码如下：

import xmltodict
from google.cloud import bigquery, storage

def load_xml_to_bq(dataset_id, table_id, gcs_xml_uri):
    # 从GCS读取XML文件
    storage_client = storage.Client()
    bucket_name, blob_path = gcs_xml_uri.replace("gs://", "").split("/", 1)
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(blob_path)
    xml_content = blob.download_as_text()
    
    # XML转字典结构
    data_dict = xmltodict.parse(xml_content)
    # 关键：根据你的XML实际结构提取数据列表
    # 示例：如果XML根节点是<orders><order>...</order></orders>，则提取order列表
    raw_data = data_dict.get("orders", {}).get("order", [])
    # 兼容单条数据场景：xmltodict会把单条节点转成字典，需手动转成列表
    if not isinstance(raw_data, list):
        raw_data = [raw_data]
    
    # 插入BigQuery
    bq_client = bigquery.Client()
    table_ref = bq_client.dataset(dataset_id).table(table_id)
    errors = bq_client.insert_rows_json(table_ref, raw_data)
    if errors:
        raise Exception(f"BigQuery插入失败: {errors}")

二、关于XML文件是否需要从GCS加载 #

不是强制要求，但在Airflow生产环境中强烈建议用GCS作为中间存储：

• 避免本地文件依赖：Airflow Worker是分布式部署的，本地文件无法跨Worker共享
• 符合GCP数据流水线最佳实践：GCS作为数据湖存储，方便数据回溯、监控
• 本地测试阶段可以直接读本地文件，但上线前必须改成GCS读取逻辑

三、你的BigQuery插入代码的问题 #

你原有的代码存在两个核心问题：

1. dec_sale_list变量未定义：这个变量应该是XML转换后得到的JSON格式数据列表，需要从转换后的字典结构中提取
2. 缺少完整的表引用逻辑：需要明确指定数据集、表的关联关系，若使用非默认GCP项目还需指定项目ID

修正后的插入逻辑已经整合到上面的函数中，核心是要保证转换后的数据结构和BigQuery表的Schema完全匹配（字段名、数据类型、嵌套结构等）。

四、整合到Airflow DAG #

下面是完整的Airflow DAG示例，包含配置、任务定义和依赖逻辑：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
from google.cloud import bigquery, storage
import xmltodict

# DAG默认参数
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

def load_xml_to_bq(**context):
    # 从DAG参数中获取配置
    dataset_id = context['params']['dataset_id']
    table_id = context['params']['table_id']
    gcs_xml_uri = context['params']['gcs_xml_uri']
    
    # GCS读取XML
    storage_client = storage.Client()
    bucket_name, blob_path = gcs_xml_uri.replace("gs://", "").split("/", 1)
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(blob_path)
    xml_content = blob.download_as_text()
    
    # XML转结构化数据
    data_dict = xmltodict.parse(xml_content)
    # 根据你的XML结构修改此处的数据提取逻辑
    raw_data = data_dict.get("root_node", {}).get("data_item", [])
    if not isinstance(raw_data, list):
        raw_data = [raw_data]
    
    # 插入BigQuery
    bq_client = bigquery.Client()
    table_ref = bq_client.dataset(dataset_id).table(table_id)
    errors = bq_client.insert_rows_json(table_ref, raw_data)
    if errors:
        raise Exception(f"插入失败: {errors}")

with DAG(
    'xml_to_bigquery_pipeline',
    default_args=default_args,
    description='从GCS加载XML文件并插入BigQuery',
    schedule_interval=timedelta(days=1),
    catchup=False,
) as dag:

    load_task = PythonOperator(
        task_id='load_xml_to_bq',
        python_callable=load_xml_to_bq,
        params={
            'dataset_id': 'your_target_dataset',
            'table_id': 'your_target_table',
            'gcs_xml_uri': 'gs://your-bucket/path/to/target.xml'
        },
        # 若使用Airflow的GCP连接，需指定conn_id
        # op_kwargs={'project_id': 'your-gcp-project-id'},
    )

load_task

五、关键注意事项 #

• 依赖安装：在Airflow环境中需安装xmltodict、google-cloud-bigquery、google-cloud-storage包
• Schema匹配：必须确保XML转换后的字段与BigQuery表的Schema完全一致，否则插入会失败
• 权限配置：Airflow Worker需要拥有GCS读取权限和BigQuery写入权限
• 数据校验：建议在插入前添加数据校验逻辑，比如检查必填字段非空、数据类型合规等

内容的提问来源于stack exchange，提问作者JohnM