Migration Hub:Migrasi dari Airflow ke DataWorks

1. Ekspor alur tugas Airflow

Untuk mengekspor metadata dari alur tugas Airflow, library Python Airflow memuat folder DAG Anda. Tool tersebut kemudian mengambil informasi mengenai DAG, tugas internalnya, serta dependensinya, lalu mengorganisasi data tersebut ke dalam file JSON untuk diekspor.

Tool migrasi mendukung ekspor alur tugas dari Airflow 2.x.

1. Lingkungan runtime

Fitur ekspor pada tool migrasi menggunakan MigrationX Airflow Reader (MigrationX) dan harus dijalankan di lingkungan Python versi 3.9.0 atau lebih baru.

Pilih salah satu lingkungan runtime yang direkomendasikan berikut untuk tool ekspor:

1. Jalankan tool di lingkungan Python yang sama dengan tempat penjadwalan Airflow berjalan, asalkan menggunakan Python 3.9.0 atau lebih baru.

2. Siapkan lingkungan Python baru dan instal versi library Python Airflow yang sama dengan lingkungan produksi Anda. Pastikan kompatibilitas antara library Airflow dan versi Python.

Jika memilih opsi kedua, ikuti langkah-langkah berikut untuk mengonfigurasi lingkungan Python baru. Contoh ini menggunakan Instance ECS Linux:

## Instal Conda dan buat lingkungan Python 3.9

# Unduh dan instal Conda
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh
sh Miniconda3-latest-Linux-x86_64.sh

# Selama instalasi, Conda akan meminta Anda menentukan direktori instalasi. Setelah instalasi, jalankan perintah berikut untuk menginisialisasinya.
cd <conda_installation_directory>
conda init
conda config --set auto_activate_base false
. ~/.bashrc

# Buat lingkungan Python 3.9.0
conda create -n airflow python=3.9.0
conda activate airflow

# Instal Airflow di lingkungan Conda (misalnya, versi 2.5.3)
pip install apache-airflow==2.5.3

2. Prosedur

2.1. Unduh paket tool ekspor airflow-workflow-parser.zip

Tautan unduh: airflow-workflow-parser

2.2. Ekstrak airflow-workflow-parser.zip

unzip airflow-workflow-parser.zip

2.3. Atur PYTHONPATH

Konfigurasikan PYTHONPATH jika Anda menjalankan tool ekspor di lingkungan Airflow. Langkah ini dapat dilewati jika tool dijalankan di lingkungan Conda.

Atur PYTHONPATH ke direktori library Python Airflow. Contohnya:

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

export AIRFLOW_HOME=/var/lib/airflow

## Catatan: Jika instans Airflow baru tidak memiliki file airflow.cfg, Anda dapat membuatnya dengan menjalankan 'airflow config list --defaults'.
export AIRFLOW_CONFIG=/var/run/cloudera-scm-agent/process/2531-airflow-AIRFLOW_SCHEDULER/airflow.cfg

2.4. Ekspor tugas Airflow

Parser mengambil semua DAG dari direktori yang ditentukan dan menghasilkan satu file DataWorks Spec (berformat JSON) untuk setiap DAG. File tersebut mendefinisikan alur tugas DataWorks.

Definisi alur tugas 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

Deskripsi parameter:

-d: Menentukan direktori tempat file DAG Airflow disimpan.

-o: Menentukan path untuk menyimpan file JSON hasil ekspor.

-m: Menentukan path file konfigurasi. Anda dapat mengonfigurasi pemetaan node dan aturan transformasi dalam file ini. Untuk informasi lebih lanjut mengenai item konfigurasi, lihat bagian 2.4.1.

Jika ekspor gagal, biasanya disebabkan oleh dependensi khusus dalam DAG. Gunakan perintah `pip` untuk menginstal dependensi yang hilang sesuai pesan error.

2.4.1. Templat konfigurasi transformasi

Tool ekspor dapat mentransformasi jenis node. Anda dapat menggunakan item konfigurasi untuk memetakan operator Airflow ke jenis node DataWorks. Contoh templat berikut menunjukkan caranya.

Templat konfigurasi untuk migrasi ke DataWorks di MaxCompute (contoh):

{
  "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"
  }
}

Templat konfigurasi untuk migrasi ke DataWorks di EMR (contoh):

{
  "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"
  }
}

Deskripsi parameter:

Buka halaman produk ruang kerja DataWorks Anda. Di panel navigasi sebelah kiri, pilih Resource Group. Lampirkan kelompok sumber daya dan dapatkan ID-nya.

2.5. Contoh

Setelah Anda mengunduh dan mengekstrak tool, Anda akan mendapatkan struktur direktori berikut:

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

Jalankan perintah.

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

Log operasional sebagai berikut:

Lihat hasil transformasi:

2.6. FAQ

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

Untuk Airflow 2.x, pastikan versi pendulum adalah 2.1.2 atau lebih lama.

Untuk Airflow 3.x, pendulum tidak diperlukan karena Airflow 3.x telah menghapus dependensi terhadap pendulum.

Solusi:

pip uninstall pendulum -y
pip install pendulum==2.1.2

2.6.2. Error dependensi saat parsing file DAG

Jika parsing gagal, biasanya karena DAG memiliki dependensi khusus. Gunakan perintah `pip` untuk menginstal dependensi yang hilang sesuai pesan error.

2. Impor ke DataWorks

Tool ekspor Airflow menghasilkan file deskripsi JSON untuk alur tugas penjadwalan yang strukturnya sesuai dengan DataWorks Spec.

Untuk mengimpor file tersebut ke DataWorks, lihat Impor DataWorks Spec kustom ke DataWorks.