E-MapReduce:Apache Airflow を使用してジョブを送信する

ステップ 1: Livy ゲートウェイとトークンを作成する

1. Livy ゲートウェイを作成して開始します。

   1. Gateways ページに移動します。

      1. EMR コンソールにログインします。

      2. 左側のナビゲーションウィンドウで、EMR Serverless > Spark を選択します。

      3. [Spark] ページで、対象のワークスペースの名前をクリックします。

      4. [EMR Serverless Spark] ページで、左側のナビゲーションウィンドウの オペレーションセンター > ゲートウェイ をクリックします。

   2. [Livy ゲートウェイ] タブをクリックします。

   3. [Livy ゲートウェイ] ページで、[Livy ゲートウェイの作成] をクリックします。

   4. [ゲートウェイの作成] ページで、[名前] (例: Livy-gateway) を入力し、[作成] をクリックします。

      必要に応じて他のパラメーターを構成できます。詳細については、「ゲートウェイ管理」をご参照ください。

   5. [Livy ゲートウェイ] ページで、作成したゲートウェイを見つけ、[アクション] 列の [開始] をクリックします。

2. トークンを作成します。

   1. [ゲートウェイ] ページで、Livy-gateway を見つけ、[アクション] 列の [トークン] をクリックします。

   2. [トークンの作成] をクリックします。

   3. [トークンの作成] ダイアログボックスで、[名前] (例: Livy-token) を入力し、[OK] をクリックします。

   4. トークンをコピーします。

      重要

      トークンが作成されたら、すぐにトークンをコピーする必要があります。ページを離れると、トークンは表示できなくなります。トークンが有効期限切れになったり、紛失したりした場合は、トークンをリセットするか、新しいトークンを作成してください。

ステップ 2: Apache Airflow を構成する

1. 次のコマンドを実行して、Apache Airflow に Apache Livy をインストールします:

   pip install apache-airflow-providers-apache-livy

2. 接続を追加します。

   UI を使用する

   Apache Airflow で [livy_default] という名前のデフォルト接続を見つけて、接続を変更します。または、Apache Airflow の Web UI で手動で接続を追加することもできます。詳細については、「UI を使用した接続の作成」をご参照ください。

   これには、次の情報が含まれます:

   • ホスト: ゲートウェイの エンドポイント を入力します。

   • スキーマ: https と入力します。

   • 追加: JSON 文字列を入力します。x-acs-spark-livy-token は、前のステップでコピーしたトークンを示します。

     {
       "x-acs-spark-livy-token": "6ac**********kfu"
     }

   CLI を使用する

   Airflow コマンドラインインターフェイス (CLI) を使用してコマンドを実行し、接続を追加します。詳細については、「接続の作成」をご参照ください。

   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",   # ゲートウェイのエンドポイント。
           "schema": "https",
           "extra": {
               "x-acs-spark-livy-token": "6ac**********kfu"  # 前のステップでコピーしたトークン。
           }
       }'

ステップ 3: DAG を構成してジョブを送信する

Apache Airflow では、有向非巡回グラフ (DAG) を使用してジョブの実行モードを宣言できます。次のサンプルコードは、Apache Airflow の LivyOperator を使用して Spark ジョブを実行する方法を示しています。

Object Storage Service (OSS) から取得した Python スクリプトファイルを実行します。

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),
}

# DAG の開始
livy_operator_sparkpi_dag = DAG(
    dag_id="livy_operator_sparkpi_dag",  # DAG の一意の ID。
    default_args=default_args,
    schedule_interval=None,
    start_date=datetime(2024, 5, 20),
    tags=['example', 'spark', 'livy'],
    catchup=False
)

# LivyOperator で livy タスクを定義する
# 必要に応じて file の値を置き換えます。
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

次の表にパラメーターを示します。

ステップ 1: Apache Airflow を構成する

1. airflow_alibaba_provider-0.0.3-py3-none-any.whl をダウンロードします。

2. Airflow の各ノードに airflow-alibaba-provider プラグインをインストールします。

   airflow-alibaba-provider プラグインは EMR Serverless Spark によって提供されます。これには、EMR Serverless Spark にジョブを送信するために使用される EmrServerlessSparkStartJobRunOperator コンポーネントが含まれています。

   pip install airflow_alibaba_provider-0.0.3-py3-none-any.whl

3. 接続を追加します。

   CLI を使用する

   Airflow CLI を使用してコマンドを実行し、接続を確立します。詳細については、「接続の作成」をご参照ください。

   airflow connections add 'emr-serverless-spark-id' \
       --conn-json '{
           "conn_type": "emr_serverless_spark",
           "extra": {
               "auth_type": "AK",  # AccessKey ペアは認証に使用されます。
               "access_key_id": "<yourAccesskeyId>",  # Alibaba Cloud アカウントの AccessKey ID。
               "access_key_secret": "<yourAccesskeyKey>",  # Alibaba Cloud アカウントの AccessKey Secret。
               "region": "<yourRegion>"
           }
       }'

   UI を使用する

   Airflow Web UI で手動で接続を作成できます。詳細については、「UI を使用した接続の作成」をご参照ください。

   [接続の追加] ページで、パラメーターを構成します。

   

   次の表にパラメーターを示します。

   パラメーター	説明
   Connection Id	接続 ID。この例では、emr-serverless-spark-id と入力します。
   Connection Type	接続タイプ。この例では、[Generic] を選択します。Generic が利用できない場合は、[Email] を選択することもできます。
   Extra	追加の構成。この例では、次の内容を入力します: { "auth_type": "AK", # AccessKey ペアは認証に使用されます。 "access_key_id": "<yourAccesskeyId>", # Alibaba Cloud アカウントの AccessKey ID。 "access_key_secret": "<yourAccesskeyKey>", # Alibaba Cloud アカウントの AccessKey Secret。 "region": "<yourRegion>" }

ステップ 2: DAG を構成する

Apache Airflow では、有向非巡回グラフ (DAG) を使用してジョブの実行モードを宣言できます。次のサンプルコードは、EmrServerlessSparkStartJobRunOperator を使用して Apache Airflow でさまざまな種類の 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

# 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

from __future__ import annotations

from datetime import datetime

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

# 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

from __future__ import annotations

from datetime import datetime

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

# 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

from __future__ import annotations

from datetime import datetime

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

# 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

次の表にパラメーターを示します。