Kirim pekerjaan menggunakan Apache Airflow - E-MapReduce

Apache Airflow adalah alat otomatisasi dan penjadwalan alur kerja yang andal, memungkinkan Anda mengelola orkestrasi, penjadwalan, serta pemantauan pipa data. EMR Serverless Spark menyediakan lingkungan komputasi tanpa server untuk pekerjaan pemrosesan data skala besar. Topik ini menjelaskan cara menggunakan Apache Airflow untuk mengotomatiskan pengiriman pekerjaan ke EMR Serverless Spark, sehingga memungkinkan penjadwalan otomatis dan manajemen pekerjaan pemrosesan data yang lebih efisien.

Informasi latar belakang

Apache Livy menggunakan REST API untuk berinteraksi dengan Spark, menyederhanakan komunikasi antara Spark dan server aplikasi. Untuk informasi selengkapnya tentang Livy API, lihat REST API.

Untuk mengirim pekerjaan dari Airflow, Anda dapat berinteraksi dengan EMR Serverless Spark menggunakan LivyOperator atau EmrServerlessSparkStartJobRunOperator. Pilih metode yang paling sesuai dengan skenario Anda.

Metode

Skenario

Metode 1: Kirim pekerjaan menggunakan LivyOperator

Pilih metode ini jika Anda ingin menggunakan operator Airflow open-source untuk mengirim pekerjaan ke EMR Serverless Spark.

Metode 2: Kirim pekerjaan menggunakan EmrServerlessSparkStartJobRunOperator

EmrServerlessSparkStartJobRunOperator adalah komponen yang disediakan oleh Alibaba Cloud EMR Serverless Spark. Komponen ini dirancang khusus untuk mengirim pekerjaan EMR Serverless Spark melalui Airflow dan terintegrasi erat dengan Directed Acyclic Graph (DAG) Airflow guna menyederhanakan orkestrasi dan penjadwalan pekerjaan.

Jika Anda memulai proyek baru atau membangun secara eksklusif di atas EMR Serverless Spark, gunakan Metode 2 untuk performa lebih tinggi dan kompleksitas operasional yang lebih rendah.

Prasyarat

Airflow telah diinstal dan sedang berjalan. Untuk informasi selengkapnya, lihat Instalasi Airflow.
Anda telah membuat ruang kerja. Untuk informasi selengkapnya, lihat Buat ruang kerja.

Catatan penggunaan

EmrServerlessSparkStartJobRunOperator saat ini tidak menghasilkan log pekerjaan. Untuk melihat log detail, login ke Konsol EMR, temukan eksekusi pekerjaan berdasarkan ID-nya, lalu periksa log pada tab Log Exploration atau di Spark UI.

Metode 1: Kirim pekerjaan menggunakan LivyOperator

Langkah 1: Buat Livy Gateway dan token

Buat dan mulai gateway.
1. Buka halaman Gateway.
  1. Login ke Konsol EMR.
  2. Pada panel navigasi kiri, pilih Spark.
  3. Pada halaman Spark, klik nama ruang kerja yang dituju.
  4. Pada halaman EMR Serverless Spark, pada panel navigasi kiri, klik O&M Center > Gateway.
2. Klik tab Livy Gateway.
3. Pada halaman Livy Gateway, klik Create Livy Gateway.
4. Pada halaman Create Gateway, masukkan Name (misalnya, Livy-gateway) lalu klik create.
  
  Sesuaikan parameter lain sesuai kebutuhan. Untuk informasi selengkapnya, lihat Kelola gateway.
5. Pada halaman Livy Gateway, temukan gateway yang telah Anda buat lalu klik START pada kolom Actions.
Buat token.
1. Pada halaman Gateway, temukan Livy-gateway lalu klik Token Management pada kolom Actions.
2. Klik Create Token.
3. Pada kotak dialog Create Token, masukkan Name (misalnya, Livy-token) lalu klik OK.
4. Salin informasi token tersebut.
  
  Penting
  Salin token segera setelah pembuatan karena tidak dapat dilihat kembali nanti. Jika token kedaluwarsa atau hilang, buat ulang atau reset token tersebut.

Langkah 2: Konfigurasikan Apache Airflow

Jalankan perintah berikut untuk menginstal penyedia Apache Livy di lingkungan Airflow Anda:
```
pip install apache-airflow-providers-apache-livy
```
Tambahkan koneksi.
UI

Di Airflow, temukan koneksi default bernama livy_default lalu ubah detailnya. Atau, Anda juga dapat menambahkan koneksi secara manual melalui antarmuka web Airflow. Untuk informasi selengkapnya, lihat Buat koneksi.

Konfigurasikan parameter berikut:
- Host: Masukkan Endpoint gateway.
- Schema: Masukkan https.
- Extra: Masukkan string JSON. x-acs-spark-livy-token adalah token yang telah Anda salin pada langkah sebelumnya.
```
{
  "x-acs-spark-livy-token": "6ac**********kfu"
}
```
CLI

Jalankan perintah yang sesuai menggunakan CLI Airflow untuk membuat koneksi. Untuk informasi selengkapnya, lihat Buat koneksi.
```
airflow connections add 'livy_default' \
    --conn-json '{
        "conn_type": "livy",
        "host": "pre-emr-spark-livy-gateway-cn-hangzhou.data.aliyun.com/api/v1/workspace/w-xxxxxxx/livycompute/lc-xxxxxxx",   # Endpoint gateway.
        "schema": "https",
        "extra": {
            "x-acs-spark-livy-token": "6ac**********kfu"  # Token yang telah Anda salin pada langkah sebelumnya.
        }
    }'
```

Langkah 3: Tulis DAG dan kirim pekerjaan

DAG (Directed Acyclic Graph) Airflow mendefinisikan cara tugas dieksekusi. Contoh berikut menunjukkan cara menggunakan LivyOperator di Airflow untuk menjalankan pekerjaan Spark.

Ambil dan jalankan file JAR dari Alibaba Cloud Object Storage Service (OSS).

from datetime import timedelta, datetime
from airflow import DAG
from airflow.providers.apache.livy.operators.livy import LivyOperator

default_args = {
    'owner': 'aliyun',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Inisialisasi DAG
livy_operator_sparkpi_dag = DAG(
    dag_id="livy_operator_sparkpi_dag",  # Pengenal unik untuk DAG.
    default_args=default_args,
    schedule_interval=None,
    start_date=datetime(2024, 5, 20),
    tags=['example', 'spark', 'livy'],
    catchup=False
)

# Definisikan tugas Livy dengan LivyOperator.
# Ganti path file dengan milik Anda sendiri.
livy_sparkpi_submit_task = LivyOperator(
    file="oss://<YourBucket>/jars/spark-examples_2.12-3.3.1.jar",
    class_name="org.apache.spark.examples.SparkPi",
    args=['1000'],
    driver_memory="1g",
    driver_cores=1,
    executor_memory="1g",
    executor_cores=2,
    num_executors=1,
    name="LivyOperator SparkPi",
    task_id="livy_sparkpi_submit_task", 
    dag=livy_operator_sparkpi_dag,
)

livy_sparkpi_submit_task

Tabel berikut menjelaskan parameter-parameter tersebut.

Parameter	Deskripsi
`dag_id`	Pengenal unik untuk DAG.
`schedule_interval`	Menentukan interval penjadwalan untuk DAG. `None` menunjukkan bahwa tugas harus dipicu secara manual.
`start_date`	Tanggal mulai DAG dapat dijadwalkan.
`tags`	Menambahkan tag ke DAG untuk memudahkan kategorisasi dan pencarian.
`catchup`	Mengontrol apakah tugas historis akan diisi ulang (backfill). Jika diatur ke `False`, tugas yang belum dieksekusi tidak akan di-backfill meskipun tanggal mulainya berada di masa lalu.
`file`	Path ke file untuk pekerjaan Spark Anda. Dalam contoh ini, merupakan path ke paket JAR yang diunggah ke Alibaba Cloud OSS. Ganti path sesuai kebutuhan. Untuk informasi cara mengunggah file, lihat Unggahan sederhana.
`class_name`	Kelas utama dalam paket JAR.
`args`	Argumen baris perintah yang diteruskan ke pekerjaan Spark.
`driver_memory` dan `driver_cores`	Ukuran memori dan jumlah core untuk driver, masing-masing.
`executor_memory` dan `executor_cores`	Ukuran memori dan jumlah core untuk setiap executor, masing-masing.
`num_executors`	Jumlah executor.
`name`	Nama pekerjaan Spark.
`task_id`	Pengenal unik untuk tugas Airflow.
`dag`	Mengaitkan tugas dengan DAG.

Metode 2: Gunakan EmrServerlessSparkStartJobRunOperator

Langkah 1: Konfigurasikan Apache Airflow

Unduh airflow_alibaba_provider-0.0.3-py3-none-any.whl.
Instal plugin airflow-alibaba-provider pada setiap node Airflow.

Plugin airflow-alibaba-provider, yang disediakan oleh tim EMR Serverless Spark, mencakup komponen EmrServerlessSparkStartJobRunOperator, yang dirancang khusus untuk mengirim pekerjaan EMR Serverless Spark.
```
pip install airflow_alibaba_provider-0.0.3-py3-none-any.whl
```

Tambahkan koneksi.

CLI

Jalankan perintah yang sesuai menggunakan CLI Airflow untuk membuat koneksi. Untuk informasi selengkapnya, lihat Buat koneksi.

airflow connections add 'emr-serverless-spark-id' \
    --conn-json '{
        "conn_type": "emr_serverless_spark",
        "extra": {
            "auth_type": "AK",  # Menentukan penggunaan AccessKey (AK) Alibaba Cloud untuk autentikasi.
            "access_key_id": "<yourAccesskeyId>",  # ID AccessKey Alibaba Cloud Anda.
            "access_key_secret": "<yourAccesskeyKey>",  # Rahasia AccessKey Alibaba Cloud Anda.
            "region": "<yourRegion>"
        }
    }'

UI

Tambahkan koneksi secara manual melalui antarmuka web Airflow. Untuk informasi selengkapnya, lihat Buat koneksi.

Pada halaman Add Connection, konfigurasikan parameter berikut.

Tabel berikut menjelaskan parameter-parameter tersebut.

Parameter	Deskripsi
Connection Id	Dalam contoh ini, nilainya adalah emr-serverless-spark-id.
Connection Type	Pilih Generic. Jika tipe ini tidak tersedia, Anda dapat memilih Email.
Extra	Masukkan konten berikut. `{ "auth_type": "AK", # Menentukan penggunaan AccessKey (AK) Alibaba Cloud untuk autentikasi. "access_key_id": "<yourAccesskeyId>", # ID AccessKey Alibaba Cloud Anda. "access_key_secret": "<yourAccesskeyKey>", # Rahasia AccessKey Alibaba Cloud Anda. "region": "<yourRegion>" }`

Langkah 2: Tulis DAG dan kirim pekerjaan

Contoh berikut menunjukkan cara menggunakan EmrServerlessSparkStartJobRunOperator di Airflow untuk menjalankan berbagai jenis pekerjaan Spark.

Pekerjaan JAR

Skema ini menunjukkan cara menggunakan tugas Airflow untuk mengirim pekerjaan Spark JAR yang telah dikompilasi ke Alibaba Cloud EMR Serverless Spark.

from __future__ import annotations

from datetime import datetime

from airflow.models.dag import DAG
from airflow_alibaba_provider.alibaba.cloud.operators.emr import EmrServerlessSparkStartJobRunOperator

# Abaikan argumen yang hilang yang disediakan oleh default_args
# mypy: disable-error-code="call-arg"

DAG_ID = "emr_spark_jar"

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2024, 5, 1),
    default_args={},
    max_active_runs=1,
    catchup=False,
) as dag:
    emr_spark_jar = EmrServerlessSparkStartJobRunOperator(
        task_id="emr_spark_jar",
        emr_serverless_spark_conn_id="emr-serverless-spark-id",
        region="cn-hangzhou",
        polling_interval=5,
        workspace_id="w-7e2f1750c6b3****",
        resource_queue_id="root_queue",
        code_type="JAR",
        name="airflow-emr-spark-jar",
        entry_point="oss://<YourBucket>/spark-resource/examples/jars/spark-examples_2.12-3.3.1.jar",
        entry_point_args=["1"],
        spark_submit_parameters="--class org.apache.spark.examples.SparkPi --conf spark.executor.cores=4 --conf spark.executor.memory=20g --conf spark.driver.cores=4 --conf spark.driver.memory=8g --conf spark.executor.instances=1",
        is_prod=True,
        engine_release_version=None
    )

    emr_spark_jar

Perintah SQL

Jalankan perintah SQL langsung dalam DAG Airflow.

from __future__ import annotations

from datetime import datetime

from airflow.models.dag import DAG
from airflow_alibaba_provider.alibaba.cloud.operators.emr import EmrServerlessSparkStartJobRunOperator

# Abaikan argumen yang hilang yang disediakan oleh default_args
# mypy: disable-error-code="call-arg"

ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
DAG_ID = "emr_spark_sql"

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2024, 5, 1),
    default_args={},
    max_active_runs=1,
    catchup=False,
) as dag:
    emr_spark_sql = EmrServerlessSparkStartJobRunOperator(
        task_id="emr_spark_sql",
        emr_serverless_spark_conn_id="emr-serverless-spark-id",
        region="cn-hangzhou",
        polling_interval=5,
        workspace_id="w-7e2f1750c6b3****",
        resource_queue_id="root_queue",
        code_type="SQL",
        name="airflow-emr-spark-sql",
        entry_point=None,
        entry_point_args=["-e","show tables;show tables;"],
        spark_submit_parameters="--class org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver --conf spark.executor.cores=4 --conf spark.executor.memory=20g --conf spark.driver.cores=4 --conf spark.driver.memory=8g --conf spark.executor.instances=1",
        is_prod=True,
        engine_release_version=None,
    )

    emr_spark_sql

File SQL dari OSS

Ambil dan jalankan file skrip SQL dari Alibaba Cloud Object Storage Service (OSS).

from __future__ import annotations

from datetime import datetime

from airflow.models.dag import DAG
from airflow_alibaba_provider.alibaba.cloud.operators.emr import EmrServerlessSparkStartJobRunOperator

# Abaikan argumen yang hilang yang disediakan oleh default_args
# mypy: disable-error-code="call-arg"

DAG_ID = "emr_spark_sql_2"

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2024, 5, 1),
    default_args={},
    max_active_runs=1,
    catchup=False,
) as dag:
    emr_spark_sql_2 = EmrServerlessSparkStartJobRunOperator(
        task_id="emr_spark_sql_2",
        emr_serverless_spark_conn_id="emr-serverless-spark-id",
        region="cn-hangzhou",
        polling_interval=5,
        workspace_id="w-ae42e9c92927****",
        resource_queue_id="root_queue",
        code_type="SQL",
        name="airflow-emr-spark-sql-2",
        entry_point="",
        entry_point_args=["-f", "oss://<YourBucket>/spark-resource/examples/sql/show_db.sql"],
        spark_submit_parameters="--class org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver --conf spark.executor.cores=4 --conf spark.executor.memory=20g --conf spark.driver.cores=4 --conf spark.driver.memory=8g --conf spark.executor.instances=1",
        is_prod=True,
        engine_release_version=None
    )

    emr_spark_sql_2

File Python dari OSS

Ambil dan jalankan file skrip Python dari Alibaba Cloud Object Storage Service (OSS).

from __future__ import annotations

from datetime import datetime

from airflow.models.dag import DAG
from airflow_alibaba_provider.alibaba.cloud.operators.emr import EmrServerlessSparkStartJobRunOperator

# Abaikan argumen yang hilang yang disediakan oleh default_args
# mypy: disable-error-code="call-arg"

DAG_ID = "emr_spark_python"

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2024, 5, 1),
    default_args={},
    max_active_runs=1,
    catchup=False,
) as dag:
    emr_spark_python = EmrServerlessSparkStartJobRunOperator(
        task_id="emr_spark_python",
        emr_serverless_spark_conn_id="emr-serverless-spark-id",
        region="cn-hangzhou",
        polling_interval=5,
        workspace_id="w-ae42e9c92927****",
        resource_queue_id="root_queue",
        code_type="PYTHON",
        name="airflow-emr-spark-python",
        entry_point="oss://<YourBucket>/spark-resource/examples/src/main/python/pi.py",
        entry_point_args=["1"],
        spark_submit_parameters="--conf spark.executor.cores=4 --conf spark.executor.memory=20g --conf spark.driver.cores=4 --conf spark.driver.memory=8g --conf spark.executor.instances=1",
        is_prod=True,
        engine_release_version=None
    )

    emr_spark_python

Tabel berikut menjelaskan parameter-parameter tersebut.

Parameter	Tipe	Deskripsi
`task_id`	`str`	Pengenal unik tugas Airflow.
`emr_serverless_spark_conn_id`	`str`	ID koneksi untuk EMR Serverless Spark.
`region`	`str`	Wilayah ruang kerja EMR Spark.
`polling_interval`	`int`	Interval (dalam detik) di mana Airflow melakukan polling status pekerjaan.
`workspace_id`	`str`	Pengenal unik ruang kerja EMR Spark.
`resource_queue_id`	`str`	ID antrian sumber daya yang digunakan oleh pekerjaan EMR Spark.
`code_type`	`str`	Jenis pekerjaan. Nilai yang valid adalah SQL, PYTHON, dan JAR. Makna parameter `entry_point` bervariasi berdasarkan jenis pekerjaan.
`name`	`str`	Nama pekerjaan EMR Spark.
`entry_point`	`str`	Tentukan lokasi file tugas startup, seperti file JAR, SQL, atau Python. Makna parameter ini bervariasi tergantung pada `code_type`.
`entry_point_args`	`List`	Daftar argumen yang diteruskan ke aplikasi Spark.
`spark_submit_parameters`	`str`	Berisi parameter tambahan untuk perintah `spark-submit`.
`is_prod`	`bool`	Menentukan lingkungan tempat tugas dijalankan. Jika diatur ke True, berarti tugas akan dijalankan di lingkungan produksi, dan `resource_queue_id` harus menentukan ID antrian sumber daya untuk lingkungan produksi, seperti root_queue.
`engine_release_version`	`str`	Versi mesin EMR Spark. Nilai default adalah "esr-2.1-native" (Spark 3.3.1, Scala 2.12, native runtime).