DMSNotebookOperator - Data Management - Alibaba Cloud ドキュメントセンター

このトピックでは、DMSNotebookOperator の設定について説明します。

概要

DMS で管理されているノートブックファイル (.ipynb) を実行します。

前提条件

ワークスペースが作成されていること。
リソースグループが作成されていること。
セッションを再利用する場合は、ノートブックセッションが作成されていることを確認してください。
新しいセッションを作成する場合は、ノートブックセッションの設定を含むテンプレートが必要です。

パラメーター

説明

file_path、run_params、profile、session_id、profile_id、cluster_id、session_name、profile_name、cluster_name の各パラメーターは Jinja テンプレートをサポートします。

パラメーター	型	必須	説明
file_path	string	はい	ノートブックファイル (.ipynb) のパスを指定します。
profile	dict	いいえ	ノートブックセッションのプロファイルを指定します。セッションを新規作成する場合、`profile`、`profile_id`、または`profile_name`のいずれか1つを指定する必要があります。 autoStopTime：セッションが自動的にリソースを解放するまでのアイドル時間を指定します。 mountPoints：データストレージの場所を指定します。形式は `[{"mntPath" : "/mnt/data*","dataPath" : "oss://test/*"},......]` です。`mntPath` はマウントパスを、`dataPath` は OSS パスを示します。 dependencies：Pypi パッケージの依存関係を指定します。 environments：環境変数を指定します。
profile_id	string	いいえ説明このパラメーターは、セッションを再利用しない場合に必須です。	profile_id：プロファイル ID を指定します。 profile_name：プロファイル名を指定します。説明いずれか一方を指定する必要があります。`profile_id` が優先されます。
profile_name	string
cluster_type	string		DMS ワークスペース内のコンピューティングクラスタータイプを指定します。有効な値は次のとおりです： cpu spark
cluster_id	string		cluster_id：DMS ワークスペース内のコンピューティングクラスター ID を指定します。 cluster_name：DMS ワークスペース内のコンピューティングクラスター名を指定します。説明いずれか一方を指定する必要があります。`cluster_id` が優先されます。
cluster_name	string
spec	string		ドライバーのリソース仕様を指定します。有効な値は次のとおりです： 1C4G：1 vCPU、4 GB 2C8G：2 vCPU、8 GB 4C16G：4 vCPU、16 GB 8C32G：8 vCPU、32 GB 16C64G：16 vCPU、64 GB
runtime_name	string		イメージ名を指定します。
session_id	string	いいえ説明このパラメーターは、セッションを再利用する場合に必須です。	再利用するセッションを指定します。既存のセッションを再利用する場合、`session_id` または `session_name` のいずれか1つを指定する必要があります。 session_id：セッション ID を指定します。 session_name：セッション名を指定します。説明いずれか一方を指定する必要があります。`session_id` が優先されます。
session_name	string	いいえ説明このパラメーターは、セッションを再利用する場合に必須です。
run_params	dict	いいえ	ノートブックファイル内の変数を置き換えるランタイムパラメーターを指定します。
timeout	int	いいえ	ノートブックファイルの実行タイムアウト期間を秒単位で指定します。
polling_interval	int	いいえ	実行ステータスをポーリングする間隔を秒単位で指定します。デフォルト値は 10 です。このパラメーターを 0 または負の数に設定すると、結果を待たずにタスクが送信されます。ポーリングには組み込みの再試行メカニズムが含まれます。

例

説明

task_id と dag は Airflow の特定のパラメーターです。詳細については、「Airflow 公式ドキュメント」をご参照ください。

from airflow import DAG
from airflow.decorators import task
from airflow.models.param import Param
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator

import json
from airflow.providers.alibabadms.cloud.operators.dms_notebook import DMSNotebookOperator

with DAG(
    "dms_notebook_test",
    params={
        "x":3
    },
) as dag:

    notebook_operator = DMSNotebookOperator(
        task_id='notebook_test_hz_name',
        profile_name='hansheng_profile.48',
        profile={},
        cluster_type='spark',
        cluster_name='spark_general2.218',
        spec='4C32G',
        runtime_name='Spark3.5_Scala2.12_Python3.9_General:1.0.9',
        file_path='/Workspace/code/default/test.ipynb',
        run_params={
            'a':"{{ params.x }}"
        },
        polling_interval=5,
        dag=dag
    )

    run_this_last = EmptyOperator(
        task_id="run_this_last",
        dag=dag,
    )

    notebook_operator >> run_this_last

if __name__ == "__main__":
    dag.test(
        run_conf={}
    )

説明

すべての DMS Airflow オペレーターは、タスクのキャンセルや自動再試行などの共通機能をサポートします。詳細については、「Airflow DMS Operator」をご参照ください。