Airflow迁移至DataWorks

2.1 下载导出工具包airflow-workflow-parser.zip：

下载链接：

2.2 解压airflow-workflow-parser.zip：

tar -zxvf airflow-workflow-parser.zip

2.3 设置PTYHONPATH

若在Airflow环境中运行导出工具，需配置PYTHONPATH；若在Conda环境中运行导出工具，则可跳过此步骤。

指定PYTHONPATH到airflow的python lib目录，例如：

export PYTHONPATH=/usr/local/lib/python3.6/site-packages

2.4 执行airflow任务导出

Parser会获取指定目录下的所有DAG，每个DAG对应生成一个DataWorks Spec文件（DataWorks任务流描述定义文件，json格式）。

DataWorks任务流描述定义 dataworks-spec

https://github.com/aliyun/dataworks-spec/tree/master

python ./parser.py -d /root/airflow/airflow-workflow/dags -o ./result -m ./flowspec-airflowV2-transformer-config.json

参数解释：

-d 表示Airflow DAG文件的存放目录；

-o 表示导出结果JSON文件的存放路径；

-m 表示配置文件路径，可在其中配置节点映射转换规则。具体转换配置项见2.4.1小节。

若执行失败，通常是由于DAG中有特殊依赖，按报错使用pip命令依次安装缺失的依赖即可。

{
  "workflowPathPrefix": "Airflow导入_V3/",
  "typeMapping": {
    "EmptyOperator": "VIRTUAL",
    "DummyOperator": "VIRTUAL",
    "ExternalTaskSensor": "VIRTUAL",
    "BashOperator": "DIDE_SHELL",
    "HiveToMySqlTransfer": "DI",
    "PrestoToMySqlTransfer": "DI",
    "PythonOperator": "PYTHON",
    "HiveOperator": "ODPS_SQL",
    "SqoopOperator": "DI",
    "SparkSqlOperator": "ODPS_SQL",
    "SparkSubmitOperator": "ODPS_SPARK",
    "SQLExecuteQueryOperator": "MySQL",
    "PostgresOperator": "Postgresql",
    "MySqlOperator": "MySQL",
    "default": "PYTHON"
  },
  "settings": {
    "workflow.converter.target.schedule.resGroupIdentifier": "Serverless_res_group_651147510078336_710919704558240"
  }
}

{
  "workflowPathPrefix": "Airflow导入_V3/",
  "typeMapping": {
    "EmptyOperator": "VIRTUAL",
    "DummyOperator": "VIRTUAL",
    "ExternalTaskSensor": "VIRTUAL",
    "BashOperator": "EMR_SHELL",
    "HiveToMySqlTransfer": "DI",
    "PrestoToMySqlTransfer": "DI",
    "PythonOperator": "PYTHON",
    "HiveOperator": "EMR_HIVE",
    "SqoopOperator": "EMR_SQOOP",
    "SparkSqlOperator": "EMR_SPARK_SQL",
    "SparkSubmitOperator": "EMR_SPARK",
    "SQLExecuteQueryOperator": "MySQL",
    "PostgresOperator": "Postgresql",
    "MySqlOperator": "MySQL",
    "default": "PYTHON"
  },
  "settings": {
    "workflow.converter.target.schedule.resGroupIdentifier": "Serverless_res_group_651147510078336_710919704558240"
  }
}

2.5 操作示例

下载工具到本地后，解压缩得到如下目录：

.
├── airflow_workflow
│   ├── common
│   ├── connections.py
│   ├── converter
│   ├── dag_parser.py
│   ├── downloader.py
│   ├── get_dags.py
│   ├── __init__.py
│   ├── miscs
│   ├── models
│   ├── patch
│   ├── __pycache__
│   └── test
├── airflow-workflow.tgz
├── dags
│   ├── example.py
│   └── __pycache__
├── flowspec-airflowV2-transformer-config.json
├── parser.py
├── README.MD
└── result

执行命令。

python ./parser.py -d /root/airflow/airflow-workflow/dags -o ./result -m ./flowspec-airflowV2-transformer-config.json

运行日志如下：

2025-05-13 15:27:05,740] {parser.py:148} INFO - Overwriting type mapping with file: ./flowspec-airflowV2-transformer-config.json
2025-05-13 15:27:05,741] {parser.py:157} INFO - Effective type mappings are:
        EmptyOperator: VIRTUAL
        DummyOperator: VIRTUAL
        ExternalTaskSensor: VIRTUAL
        BashOperator: ODPS_SQL
        HiveToMySqlTransfer: DI
        PrestoToMySqlTransfer: DI
        PythonOperator: PYTHON
        BranchPythonOperator: CONTROLLER_BRANCH
        HiveOperator: EMR_HIVE
        SqoopOperator: EMR_SQOOP
        SparkSqlOperator: EMR_SPARK_SQL
        SparkSubmitOperator: EMR_SPARK
        SQLExecuteQueryOperator: MySQL
        PostgresOperator: Postgresql
        MySqlOperator: MySQL
        default: PYTHON
2025-05-13 15:27:05,741] {parser.py:160} INFO - Also setting workflow path prefix to Airflow导入_V3/
2025-05-13 15:27:05,741] {parser.py:164} INFO - Also overwriting settings with {'workflow.converter.target.schedule.resGroupIdentifier': 'Serverless_res_group_651147510078336_710919704558240'}
2025-05-13 15:27:05,741] {parser.py:169} INFO - Effective workflow path prefix is: Airflow导入_V3/
2025-05-13 15:27:05,742] {parser.py:48} INFO - conf: /root/airflow
2025-05-13 15:27:05,742] {dagbag.py:500} INFO - Filling up the DagBag from /root/airflow/airflow-workflow/dags
2025-05-13 15:27:05,751] {dag_parser.py:13} INFO - Init DagBag success
2025-05-13 15:27:05,753] {dag_converter.py:27} INFO - Converting DAG tutorial to DataWorks workflow tutorial
/root/airflow/airflow-workflow/airflow_workflow/converter/task_converter.py:37 DeprecationWarning: DAG.full_filepath is deprecated in favour of fileloc
2025-05-13 15:27:05,755] {dag_converter.py:62} INFO - Finished converting DAG tutorial
2025-05-13 15:27:05,756] {parser.py:187} INFO - Finished parsing Airflow DAGs.

查看转换结果：

(airflow) [root@iZbp13j3jwx9rjos76cb6tZ airflow-workflow]# ls result/
tutorial.json

(airflow) [root@iZbp13j3jwx9rjos76cb6tZ airflow-workflow]# cat result/tutorial.json
{
  "version": "1.1.0",
  "kind": "CycleWorkflow",
  "spec": {
    "nodes": [],
    "workflows": [
      {
        "id": "484353000179523",
        "outputs": {
          "nodeOutputs": [
            {
              "artifactType": "NodeOutput",
              "data": "484353000179523",
              "refTableName": "tutorial"
            }
          ]
        },
        "nodes": [
          {
            "id": "2263957773817806",
            "name": "print_date",
            ...
            ...
          }
        ]
      }
    ]
  }
}

2.6 常见问题

pip uninstall pendulum -y
pip install pendulum==2.1.2