DMS による Lindorm Spark ジョブのオーケストレーションとスケジュール - Lindorm - Alibaba Cloud - Lindorm

Lindorm が提供する Lindorm 分散処理システム (LDPS) では、データ管理 (DMS) のタスクオーケストレーション機能を利用して、Lindorm Spark ジョブのスケジューリング、およびその実行履歴とログの表示が可能です。LDPS は、データプロダクション、インタラクティブ分析、機械学習、グラフコンピューティングなどのシナリオにおけるコンピューティング要件を満たすことができます。このトピックでは、DMS を使用して Lindorm Spark ジョブを管理する方法について説明します。

前提条件

DMS が有効化されていること。
Lindorm インスタンスで LDPS が有効化されていること。詳細については、「サービスの有効化」をご参照ください。
ジョブが開発済みであること。詳細については、JAR ジョブ開発プラクティスまたはPython ジョブ開発：実践ガイドをご参照ください。
開発したジョブが HDFS または Object Storage Service (OSS) にアップロードされていること。詳細については、「コンソールを使用したファイルのアップロード」をご参照ください。

Lindorm Spark タスクフローの作成

DMS コンソール V5.0 にログインします。
タスクオーケストレーション ページに移動します。
- シンプルモード：
  1. シーンガイダンス セクションで、[データ転送と処理 (DTS)] をクリックします。
  2. ページの右側にある [データ処理] セクションの タスクオーケストレーション をクリックします。
- 通常モード：上部メニューで、[DTS] > データ開発 > タスクオーケストレーション を選択します。
タスクオーケストレーション ページで、新しいタスクフロー をクリックします。
新しいタスクフロー ダイアログボックスで、[タスクフロー名] と説明を指定し、OK をクリックします。
左側の タスクタイプ セクションで、[Lindorm Spark] ノードをキャンバスにドラッグし、ノード同士を接続してノード間の依存関係を指定します。

[Lindorm Spark] ノードを設定します。

[Lindorm Spark] ノードをダブルクリックするか、[Lindorm Spark] ノードをクリックしてからアイコンをクリックします。

表示されたページで、実行するジョブの基本パラメータとカスタムパラメータを設定します。

基本設定 セクションで、基本パラメータを設定します。基本パラメータを次の表に示します。

パラメータ	説明
リージョン	Lindorm インスタンスがデプロイされているリージョンを選択します。
Lindorm インスタンス	Lindorm インスタンスの ID を選択します。
タスクタイプ	Spark ジョブのタイプを選択します。次のジョブタイプがサポートされています： JAR Python SQL

ジョブ設定 セクションで、実行するジョブのカスタムパラメータを設定します。以降のセクションでは、さまざまなタイプの Spark ジョブの設定テンプレートとカスタムパラメータについて説明します。

JAR Spark ジョブの設定テンプレートとカスタムパラメータは次のとおりです：

{
  "mainResource" : "oss://path/to/your/file.jar",
  "mainClass" : "path.to.main.class",
  "args" : [ "arg1", "arg2" ],
  "configs" : {
    "spark.hadoop.fs.oss.endpoint" : "",
    "spark.hadoop.fs.oss.accessKeyId" : "",
    "spark.hadoop.fs.oss.accessKeySecret" : "",
    "spark.hadoop.fs.oss.impl" : "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem",
    "spark.sql.shuffle.partitions" : "20"
  }
}

パラメータ	タイプ	必須	説明	例
mainResource	String	はい	JAR パッケージが格納されている HDFS または OSS のパス。	JAR パッケージが格納されている HDFS パス：hdfs:///path/spark-examples_2.12-3.1.1.jar JAR パッケージが格納されている OSS パス：oss://testBucketName/path/spark-examples_2.12-3.1.1.jar
mainClass	String	はい	JAR ジョブ内のプログラムのエントリポイントとして使用されるクラス。	com.aliyun.ldspark.SparkPi
args	Array	いいえ	mainClass パラメータに渡されるパラメータ。	["arg1", "arg2"]
configs	Json	いいえ	Spark ジョブのシステムパラメータ。ジョブを OSS にアップロードする場合は、configs で次のパラメータを設定する必要があります： spark.hadoop.fs.oss.endpoint：Spark ジョブが格納されている OSS のパス。 spark.hadoop.fs.oss.accessKeyId：OSS へのアクセスに使用される AccessKey ID。コンソールで AccessKey ID を取得できます。詳細については、AccessKeyペアの取得をご参照ください。 spark.hadoop.fs.oss.accessKeySecret：OSS へのアクセスに使用される AccessKey Secret。コンソールで AccessKey Secret を取得できます。詳細については、AccessKeyペアの取得をご参照ください。 spark.hadoop.fs.oss.impl：OSS へのアクセスに使用されるクラス。値を org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem に設定します。	{ "spark.sql.shuffle.partitions": "200"}

Python Spark ジョブの設定テンプレートとカスタムパラメータは次のとおりです：

{
  "mainResource" : "oss://path/to/your/file.py",
  "args" : [ "arg1", "arg2" ],
  "configs" : {
    "spark.hadoop.fs.oss.endpoint" : "",
    "spark.hadoop.fs.oss.accessKeyId" : "",
    "spark.hadoop.fs.oss.accessKeySecret" : "",
    "spark.hadoop.fs.oss.impl" : "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem",
    "spark.submit.pyFiles" : "oss://path/to/your/project_file.py,oss://path/to/your/project_module.zip",
    "spark.archives" : "oss://path/to/your/environment.tar.gz#environment",
    "spark.sql.shuffle.partitions" : "20"
  }
}

パラメータ	タイプ	必須	説明	例
mainResource	String	はい	Python ファイルが格納されている OSS または HDFS のパス。	Python ファイルが格納されている OSS パス：oss://testBucketName/path/spark-examples.py Python ファイルが格納されている HDFS パス：hdfs:///path/spark-examples.py
args	Array	いいえ	Python スクリプトに渡されるパラメータ。	["arg1", "arg2"]
configs	Json	いいえ	Spark ジョブのシステムパラメータ。ジョブを OSS にアップロードする場合は、configs で次のパラメータを設定する必要があります： spark.hadoop.fs.oss.endpoint：Spark ジョブが格納されている OSS のパス。 spark.hadoop.fs.oss.accessKeyId：OSS へのアクセスに使用される AccessKey ID。コンソールで AccessKey ID を取得できます。詳細については、AccessKeyペアの取得をご参照ください。 spark.hadoop.fs.oss.accessKeySecret：OSS へのアクセスに使用される AccessKey Secret。コンソールで AccessKey Secret を取得できます。詳細については、AccessKeyペアの取得をご参照ください。 spark.hadoop.fs.oss.impl：OSS へのアクセスに使用されるクラス。値を org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem に設定します。	{"spark.sql.shuffle.partitions": "200"}

SQL ジョブの設定テンプレートとカスタムパラメータは次のとおりです：

{
  "mainResource" : "oss://path/to/your/file.sql",
  "configs" : {
    "spark.hadoop.fs.oss.endpoint" : "",
    "spark.hadoop.fs.oss.accessKeyId" : "",
    "spark.hadoop.fs.oss.accessKeySecret" : "",
    "spark.hadoop.fs.oss.impl" : "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem",
    "spark.sql.shuffle.partitions" : "20"
  }
}

パラメータ	タイプ	必須	説明	例
mainResource	String	はい	SQL ファイルが格納されている OSS または HDFS のパス。	SQL ファイルが格納されている OSS パス：oss://testBucketName/path/spark-examples.sql SQL ファイルが格納されている HDFS パス：hdfs:///path/spark-examples.sql
configs	Json	いいえ	SQL ジョブのその他の設定。	{ "spark.executor.memory" : "8g"}

上記の設定が完了したら、左上隅の 試運転 をクリックして、ジョブが期待どおりに実行されるかどうかを確認します。

タスクフローを公開します。すべてのノードを設定したら、現在のタスクフローページの左上隅にある パブリッシュ をクリックします。

タスクフローの公開履歴とログの表示

タスクオーケストレーション ページで、公開履歴とログを表示したいタスクフローの名前をクリックします。
表示されたページで、右上隅の オペレーション & メンテナンス (o & m) に移動します をクリックします。
タスクフローの公開履歴とログを表示します。
- タスクフローの公開履歴を表示します。タスクフロー情報 ページで 公開リスト タブをクリックして、タスクフローの公開履歴を表示します。公開済みのタスクリストには、[バージョン ID]、[公開者]、[公開日時]、[備考]、[操作] の各列があります。[操作] 列で [詳細] または [DAG] をクリックすると、バージョンの公開詳細を表示できます。
- タスクフローのログを表示します。
  1. 実行中のレコード タブで、左上隅のドロップダウンリストから タイミングトリガー または 手動トリガー を選択して、タスクフロー内のすべてのノードの詳細を表示します。[実行履歴] タブで実行記録を展開すると、各サブタスクノードの実行ステータス、所要時間、実行ログが表示されます。
  2. 表示したいノードの行にある詳細をクリックします。次に、Lindorm Spark ジョブの送信ログを表示し、ノードのジョブ ID と SparkUI を取得します。
    
    説明
    ジョブの送信に失敗した場合は、チケットを提出する際にジョブ ID と SparkUI を提供してください。

詳細設定

説明

DMS コンソールで Lindorm Spark タスクフローを設定できます。Lindorm Spark タスクフローを設定した後、タスクフローを再公開する必要があります。

スケジューリング設定の構成

ビジネス要件に基づいてスケジューリングポリシーを設定できます。Lindorm Spark タスクフローは、スケジューリングポリシーに基づいて自動的に実行されます。次の手順でスケジューリングポリシーを設定します。

タスクオーケストレーション ページで、スケジューリングポリシーを設定するタスクフローの名前をクリックします。
表示されたページの左下隅にある タスクフロー情報 をクリックします。

右側の [スケジューリング設定] セクションで、スケジューリングの有効化 をオンにし、スケジューリングポリシーを設定します。設定可能なパラメーターを次の表に示します。

パラメーター	説明
スケジューリングタイプ	スケジューリングタイプを選択します。周期的スケジューリング：週に 1 回など、定期的にタスクを実行します。 1回限りのスケジューリング：指定した時間にタスクを 1 回実行します。特定の実行時間を設定するだけです。
有効期間	スケジューリング周期が有効になる期間を選択します。デフォルト値は 1970-01-01 から 9999-01-01 で、スケジューリングが常に有効であることを意味します。
スケジューリング周期	スケジューリング周期を選択します。時間：指定した時間に基づいてタスクを実行します。時間指定スケジューリングの設定が必要です。日：タスクを 1 日に 1 回実行します。毎日の実行時刻を設定する必要があります。週：指定された各曜日にタスクを 1 回実行します。実行する曜日と時刻を設定する必要があります。月：指定された各日にタスクを 1 回実行します。実行する日付と時刻を設定する必要があります。
時間指定スケジューリング	2 つの時間指定スケジューリング方法があります。固定間隔スケジューリング：開始時刻：タスクの実行開始時刻。間隔：タスク実行の間隔 (時間単位)。終了時刻：タスクの実行終了時刻。たとえば、開始時刻を 00:00、間隔を 6 時間、終了時刻を 20:59 に設定した場合、タスクは 00:00、06:00、12:00、18:00 に実行されます。指定時刻スケジューリング：タスクを実行する特定の時刻を選択します。たとえば、0 時と 5 時を選択した場合、タスクは 00:00 と 05:00 に実行されます。
指定時刻	スケジューリング周期が「週」の場合、タスクを実行する曜日を選択します。複数選択が可能です。スケジューリング周期が「月」の場合、タスクを実行する日付を選択します。複数選択が可能です。
特定時刻	タスクフロー実行の特定時刻を設定します。たとえば、時刻を 02:55 に設定した場合、タスクは指定された日の 02:55 に実行されます。
Cron 式	手動での設定は不要です。システムは、設定された周期と特定時刻に基づいて Cron 式を自動的に生成します。

例：タスクフローを毎日 00:00 と 12:00 にスケジュールしたい場合は、次のパラメーターを設定してスケジューリングポリシーを構成します。

[スケジューリングタイプ] を [周期的スケジューリング] に設定します。
[スケジューリング周期] ドロップダウンリストから [時間] を選択します。
定期実行 フィールドで、[指定時刻] を選択します。[指定時刻] ドロップダウンリストから [0時] と [12時] を選択します。

変数の設定

周期的スケジューリングが有効になっているタスクフローでは、実行するジョブの時間変数を設定できます。たとえば、ノードに時間変数 bizdate を設定できます。この時間変数は、タスクが実行される時点の前日を示します。時間変数は、次の手順で設定します。

現在のタスクフローページで、[Lindorm Spark] ノードをダブルクリックするか、[Lindorm Spark] ノードをクリックしてアイコンをクリックします。
右側のナビゲーションペインで、変数設定 をクリックします。
ノード変数 または [タスクフロー変数] タブで、変数を追加します。
ジョブ設定 セクションで、変数を使用します。他の変数については、「変数」をご参照ください。

ジョブパラメーターで定義した変数は、${変数名} の形式で参照します。

通知の管理

タスクフローの通知機能を有効にすると、システムはタスクフローの実行結果に基づいて通知メッセージを送信します。通知機能は、次の手順で有効にします。

現在のタスクフローページの左下隅にある 通知設定 をクリックします。
ビジネス要件に基づいて、次のいずれかの通知スイッチをオンにします。
- 基本通知
  - 成功通知：タスクフローが正常に実行された場合、システムは通知メッセージを送信します。
  - 失敗通知：タスクフローの実行に失敗した場合、システムは通知メッセージを送信します。
- タイムアウト通知：タスクフローがタイムアウトした場合、システムは通知メッセージを送信します。
- アラート通知：タスクが開始される直前に、システムは通知メッセージを送信します。
(オプション)：メッセージの受信者を設定します。メッセージ受信者の設定方法の詳細については、「通知管理」をご参照ください。

SQL ステートメントの実行

DMS コンソール V5.0 にログインします。
[ホーム] タブをクリックします。
左側のナビゲーションペインで、アイコンをクリックしてインスタンスを作成します。
[インスタンスの追加] ダイアログボックスの [NoSQL データベース] セクションで、[Lindorm_Compute] を選択します。
インスタンスの[インスタンスリージョン]、[インスタンス ID]、[データベースアカウント]、および[データベースパスワード]を指定し、[送信]をクリックします。
表示されたダイアログボックスで [送信] をクリックすると、SQLConsole に移動します。
[SQLConsole] タブで、実行する SQL 文を入力し、[実行] をクリックします。

リファレンス

DMS のタスクオーケストレーション機能の詳細については、「概要」をご参照ください。