使用Airflow Python Operator复制本地文件报错：FileNotFoundError

我编写了一个Airflow DAG，通过Python Operator实现本地系统文件夹间的文件复制功能，代码如下，运行时出现FileNotFoundError: [Errno 2] No such file or directory错误，需要排查解决。

原代码 #

文件复制函数 #

import os
import shutil
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta


def copy_tdsx_hyper_template_if_not_exists(source_files, destination_folder):
    for source_file in source_files:
        file_name = os.path.basename(source_file)
        destination_path = os.path.join(destination_folder, file_name)
        if not os.path.exists(destination_path):
            shutil.copyfile(source_file, destination_path)
            print(f"Template File '{file_name}' copied successfully.")

DAG定义 #

default_args = {
    'owner': 'test',
    'retries': 5,
    'retry_delay': timedelta(minutes=2)
}

with DAG(
    default_args=default_args,
    dag_id="HyperAPI_extracts_dag_test",
    description="DAG Operation Using Python Operator",
    start_date=datetime(2023, 5, 28),
    schedule='@daily'
) as dag:
    # Define the task
    parent_dir = 'C:\\localfolder\\'

Tableau_extract_name = 'TestExtract'

my_source_files = [
    parent_dir + 'Tdsx_Hyper_Templates\\\\' + Tableau_extract_name + '.hyper',
    parent_dir + 'Tdsx_Hyper_Templates\\\\' + Tableau_extract_name + '.tdsx'
]
my_destination_folder = parent_dir

copy_task = PythonOperator(
    task_id='copy_tdsx_hyper_temp_task',
    python_callable=copy_tdsx_hyper_template_if_not_exists,
    op_args=[my_source_files, my_destination_folder],
    dag=dag,
)

# Set task dependencies
copy_task 

排查与解决方法 #

• 路径转义与拼接错误：Windows路径手动转义容易出错，建议使用原始字符串或os.path.join()自动处理路径分隔符，避免手动写\\\\的错误。示例修改：

  parent_dir = r'C:\localfolder\'
  my_source_files = [
      os.path.join(parent_dir, 'Tdsx_Hyper_Templates', f"{Tableau_extract_name}.hyper"),
      os.path.join(parent_dir, 'Tdsx_Hyper_Templates', f"{Tableau_extract_name}.tdsx")
  ]
  

• 源文件/目录不存在：先确认以下路径是否真实存在：
  • C:\localfolder\Tdsx_Hyper_Templates\TestExtract.hyper
  • C:\localfolder\Tdsx_Hyper_Templates\TestExtract.tdsx
    可在函数中增加源文件存在性检查，明确错误来源：

  def copy_tdsx_hyper_template_if_not_exists(source_files, destination_folder):
      for source_file in source_files:
          if not os.path.exists(source_file):
              print(f"Error: Source file '{source_file}' does not exist.")
              continue
          file_name = os.path.basename(source_file)
          destination_path = os.path.join(destination_folder, file_name)
          if not os.path.exists(destination_path):
              shutil.copyfile(source_file, destination_path)
              print(f"Template File '{file_name}' copied successfully.")
  

• Airflow运行环境权限/挂载问题：
  • 若Airflow运行在Docker等容器中，需确保本地C:\localfolder目录已挂载到容器内对应路径，否则容器无法访问本地文件。
  • 若为本地运行Airflow，检查运行Airflow的用户是否拥有C:\localfolder目录的读写权限。
• DAG代码作用域优化：原代码中Tableau_extract_name等变量定义在DAG代码块外部，建议移到with DAG(...) as dag:内部，保持代码结构清晰：

  with DAG(...) as dag:
      parent_dir = r'C:\localfolder\'
      Tableau_extract_name = 'TestExtract'
      my_source_files = [
          os.path.join(parent_dir, 'Tdsx_Hyper_Templates', f"{Tableau_extract_name}.hyper"),
          os.path.join(parent_dir, 'Tdsx_Hyper_Templates', f"{Tableau_extract_name}.tdsx")
      ]
      my_destination_folder = parent_dir
      # 后续任务定义...
  

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