Migration Hub:Migrate Airflow DAGs to DataWorks

1. Export Airflow task flows

The export process loads your DAG folder using the Airflow Python library, extracts DAG metadata such as tasks and dependencies, and outputs a JSON file.

The migration tool supports exporting task flows from Airflow 2.x.

1. Runtime environment

The export feature uses MigrationX Airflow Reader and requires Python 3.9.0 or later.

Recommended runtime environments:

1. Use the same Python environment where Airflow runs (Python 3.9.0 or later).

2. Create a new Python environment with the same Airflow library version as production. Ensure Airflow and Python version compatibility.

Set up a new Python environment (Linux ECS example):

## Install Conda and create a Python 3.9 environment

# Download and install Conda
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
sh Miniconda3-latest-Linux-x86_64.sh

# During installation, Conda asks you to specify an installation directory. After installation, run the following commands to initialize it.
cd <conda_installation_directory>
conda init
conda config --set auto_activate_base false
. ~/.bashrc

# Create a Python 3.9.0 environment
conda create -n airflow python=3.9.0
conda activate airflow

# Install Airflow in the Conda environment (for example, version 2.5.3)
pip install apache-airflow==2.5.3

2. Procedure

2.1. Download the export tool package airflow-workflow-parser.zip

Download link: airflow-workflow-parser

2.2. Decompress airflow-workflow-parser.zip

unzip airflow-workflow-parser.zip

2.3. Set PYTHONPATH

Set PYTHONPATH if you run the export tool in an Airflow environment. Skip this step for Conda environments.

Point PYTHONPATH to the Airflow Python library directory:

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

export AIRFLOW_HOME=/var/lib/airflow

## Note: If a new Airflow instance does not have an airflow.cfg file, you can generate one by running 'airflow config list --defaults'.
export AIRFLOW_CONFIG=/var/run/cloudera-scm-agent/process/2531-airflow-AIRFLOW_SCHEDULER/airflow.cfg

2.4. Export Airflow tasks

The parser scans the specified directory and generates a DataWorks Spec JSON file for each DAG.

DataWorks task flow definition: 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

Parameters:

-d: Directory containing the Airflow DAG files.

-o: Output path for exported JSON files.

-m: Configuration file path for node mapping and transform rules (section 2.4.1).

Export failures are typically caused by missing DAG dependencies. Install them with `pip` as indicated by the error messages.

2.4.1. Transform configuration template

The configuration file maps Airflow operators to DataWorks node types.

MaxCompute migration template:

{
  "workflowPathPrefix": "Airflow_Import_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"
  }
}

EMR migration template:

{
  "workflowPathPrefix": "Airflow_Import_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"
  }
}

Parameters:

Parameter Name	Description
workflowPathPrefix	Storage path for imported task flows in DataWorks.
typeMapping	Maps Airflow operators to DataWorks node types. DataWorks node type enumeration: https://github.com/aliyun/dataworks-spec/blob/b0f4a4fd769215d5f81c0bbe990addd7498df5f4/spec/src/main/java/com/aliyun/dataworks/common/spec/domain/dw/types/CodeProgramType.java#L180
workflow.converter.target.schedule.resGroupIdentifier	The ID of the DataWorks resource group.

On the product page of your DataWorks workspace, go to Resource Group in the left pane to attach a resource group and obtain its ID.

2.5. Example

The decompressed tool has the following structure:

.
├── 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

Run the command.

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

Output logs:

View the transform results:

2.6. FAQ

2.6.1. Error: TIMEZONE = pendulum.tz.timezone("UTC") TypeError: 'module' object is not callable

For Airflow 2.x, ensure that the pendulum version is 2.1.2 or earlier.

Airflow 3.x does not require pendulum.

Solution:

pip uninstall pendulum -y
pip install pendulum==2.1.2

2.6.2. DAG file parsing dependency error

Parsing failures are typically caused by missing DAG dependencies. Install them with `pip` as indicated by the error messages.

2. Import into DataWorks

The export tool produces a DataWorks Spec-compliant JSON file for each task flow.

Import a custom DataWorks Spec into DataWorks.