如何在Airflow的BashOperator中xcom_push解压.gz文件后的文件名？

这个需求很常见，利用Airflow内置的XCom机制结合BashOperator的特性就能轻松实现，我给你分步拆解具体操作：

1. 修改BashOperator，推送解压后的文件名到XCom #

BashOperator默认会把bash命令执行后的**标准输出（stdout）**作为return_value自动推送到XCom（前提是do_xcom_push=True，这是默认配置）。所以我们只需要在解压命令后追加一行输出文件名的命令，就能让XCom捕获到这个值。

修改后的任务代码如下：

gzip_file = BashOperator(
    task_id="gzip_file",
    # 先执行解压，成功后输出解压后的文件名
    bash_command="gzip -d archive_name.csv.gz && echo archive_name.csv",
    dag=dag
)

这样一来，当gzip -d命令成功执行后，echo archive_name.csv的输出会被XCom存储起来，供后续任务拉取。

2. 在其他任务中拉取XCom中的文件名 #

假设你用PythonOperator来处理后续逻辑，可以通过任务上下文（context）中的ti（TaskInstance）对象拉取XCom的值。

示例代码：

from airflow.operators.python import PythonOperator

def process_unzipped_file(**context):
    # 从上下文获取TaskInstance对象
    ti = context["ti"]
    # 拉取gzip_file任务推送的文件名
    unzipped_filename = ti.xcom_pull(task_ids="gzip_file")
    
    print(f"成功获取到解压后的文件名：{unzipped_filename}")
    # 这里可以添加你的业务逻辑，比如读取文件、数据处理等

# 定义后续处理任务
process_file_task = PythonOperator(
    task_id="process_unzipped_file",
    python_callable=process_unzipped_file,
    provide_context=True,  # 传递上下文给Python函数
    dag=dag
)

# 设置任务依赖：先解压再处理
gzip_file >> process_file_task

3. 处理动态文件名的场景（可选） #

如果你的压缩文件名不是固定的（比如带日期后缀），可以用Airflow的模板变量来动态生成命令和推送的文件名。比如通过params传递动态名称：

gzip_file = BashOperator(
    task_id="gzip_file",
    bash_command="gzip -d {{ params.archive_name }}.csv.gz && echo {{ params.archive_name }}.csv",
    params={"archive_name": f"archive_{{ ds_nodash }}"},  # 用日期变量生成动态名称
    dag=dag
)

这里的{{ ds_nodash }}是Airflow的内置模板变量，会自动替换为当前执行日期的无横杠格式（比如20240520）。

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