Airflow中BashOperator能否访问PythonOperator创建的目录文件？

可以访问，但需要满足两个核心前提：

• 共享执行环境：两个任务必须运行在同一个Airflow Worker节点（或同一个Kubernetes Pod，若使用KubernetesExecutor）。如果任务被调度到不同Worker，本地目录是相互隔离的，此时需要改用共享存储（比如挂载GCS到本地路径、使用NFS等）来存放文件，而非本地临时目录。
• 使用绝对路径：PythonOperator中创建目录和保存文件时必须用绝对路径，避免相对路径导致BashOperator找不到位置。同时可以通过XCom将目录路径传递给BashOperator，不用硬编码路径。

举个简单的实现示例：

1. PythonOperator代码片段：

def filter_and_download_from_gcs(**context):
    import os
    from google.cloud import storage

    # 定义绝对目标目录
    target_dir = "/opt/airflow/filtered_target_files"
    os.makedirs(target_dir, exist_ok=True)

    # 文件名校验、从GCS下载符合条件文件到target_dir的逻辑
    client = storage.Client()
    bucket = client.get_bucket("your-gcs-bucket")
    # filename_list为预定义的目标文件名列表
    for filename in filename_list:
        blob = bucket.blob(filename)
        blob.download_to_filename(os.path.join(target_dir, filename))
    
    # 将目录路径通过XCom传递给后续任务
    context["task_instance"].xcom_push(key="target_dir", value=target_dir)

2. BashOperator代码片段：

BashOperator(
    task_id="process_files_with_grep_sed",
    bash_command="""
        grep '需要提取的行内容' {{ ti.xcom_pull(task_ids='filter_download_task', key='target_dir') }}/*.txt | \
        sed 's/原内容/替换内容/' > /opt/airflow/processed_results/final_output.txt
    """,
    task_concurrency=1
)

额外注意：确保Airflow运行用户对目标目录有读写权限，默认情况下PythonOperator和BashOperator使用同一个用户（如airflow），权限不会有问题；若自定义了运行用户，需提前配置好目录权限。

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