如何在Apache Airflow 2.3.4中通过PythonOperator高效复用外部Python代码？

方法一：把代码仓库路径加入Python路径 #

Airflow的Python解释器默认不会识别/home/sources这个路径，所以得先把它加到Python的sys.path里，这样就能直接导入仓库里的模块。

具体操作步骤 #

• 在你的DAG文件最开头，添加路径配置：

  import sys
  sys.path.append('/home/sources')
  
  from main import main  # 直接导入仓库里的main函数
  

• 接着在PythonOperator里调用这个函数，还能通过op_kwargs传参：

  from airflow import DAG
  from airflow.operators.python import PythonOperator
  from datetime import datetime
  
  default_args = {
      'owner': 'airflow',
      'start_date': datetime(2024, 1, 1),
  }
  
  with DAG('reuse_external_code_dag',
           default_args=default_args,
           schedule_interval='@daily',
           catchup=False) as dag:
  
      run_main_task = PythonOperator(
          task_id='run_main_function',
          python_callable=main,
          op_kwargs={'param1': 'value1', 'param2': 123}  # 给main函数传的参数
      )
  

方法二：把代码仓库作为Python包安装（更规范） #

你的代码仓库已经有__init__.py，属于标准Python包，可以用可编辑模式安装，这样代码改了之后不用重新安装，Airflow就能直接用最新版本。

具体操作 #

• 在Airflow运行的环境里执行安装命令：

  pip install -e /home/sources
  

• 之后在DAG文件里直接导入就行，不用改sys.path：

  from main import main  # 直接导入安装好的包模块
  

• PythonOperator的用法和上面一样，直接传python_callable=main再加参数就行。

注意事项 #

• 要保证Airflow运行的用户对/home/sources目录有读权限，不然会出现导入失败的问题。
• 用方法一的话，路径添加操作要放在DAG文件最开头，避免导入顺序出错。
• 方法二更适合长期维护的代码仓库，能避免路径管理混乱，也符合Python包的规范。

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