ADB Spark の使用 - DataWorks - Alibaba Cloud ドキュメントセンター

背景情報

ADB Spark は、AnalyticDB サービス内のコンピュートエンジンであり、大規模な Apache Spark データ処理タスクを実行するように設計されています。リアルタイムのデータ分析、複雑なクエリ、機械学習アプリケーションをサポートします。Java、Scala、Python をサポートしているため、開発が簡素化され、パフォーマンスとコストを最適化するために自動的にスケーリングされます。関連する Jar または .py ファイルをアップロードしてタスクを設定できます。ADB Spark は、膨大なデータセットを効率的に処理し、リアルタイムのインサイトを得る必要があるさまざまな業界に最適です。これにより、企業はデータから貴重な情報を抽出し、ビジネスの成長を促進できます。

前提条件

AnalyticDB for MySQL の前提条件：

DataWorks ワークスペースと同じリージョンに AnalyticDB for MySQL の Basic Edition のクラスターを作成済みであること。詳細については、「クラスターの作成」をご参照ください。
AnalyticDB for MySQL クラスターでジョブリソースグループを設定済みであること。詳細については、「ジョブリソースグループの作成」をご参照ください。

説明
DataWorks を使用して Spark アプリケーションを開発する場合は、ジョブリソースグループを作成する必要があります。
OSS をストレージに使用する場合、OSS バケットが AnalyticDB for MySQL クラスターと同じリージョンにあることを確認してください。

DataWorks の前提条件：

データ開発 (DataStudio) (新版) パブリックプレビューへの参加 オプションが選択され、リソースグループがアタッチされたワークスペースがあること。詳細については、「ワークスペースの作成」をご参照ください。
リソースグループは AnalyticDB for MySQL クラスターと同じ VPC に配置し、リソースグループの IP アドレスを AnalyticDB for MySQL クラスターのホワイトリストに追加する必要があります。詳細については、「ホワイトリストの設定」をご参照ください。
AnalyticDB for MySQL クラスターインスタンスをコンピュートエンジンとして DataWorks に追加済みであること。コンピュートエンジンのタイプは AnalyticDB for Spark である必要があります。DataWorks は、リソースグループを使用してコンピュートエンジンの接続性をテストします。詳細については、「コンピュートエンジンのバインド」をご参照ください。
ワークフローフォルダーを作成済みであること。詳細については、「ワークフローフォルダー」をご参照ください。
ADB Spark ノードを作成済みであること。詳細については、「ワークフローのノード作成」をご参照ください。

ステップ 1：ADB Spark ノードの開発

ADB Spark ノードエディターでは、選択した言語に応じて、サンプルの spark-examples_2.12-3.2.0.jar パッケージまたは spark_oss.py ファイルを使用してノードのコンテンツを設定できます。ノードコンテンツの開発方法の詳細については、「spark-submit コマンドラインツールを使用した Spark アプリケーションの開発」をご参照ください。

Java と Scala

JAR ファイルの準備

サンプルの JAR パッケージを OSS にアップロードする必要があります。ノードがタスクを実行するには、このパッケージが必要です。

サンプルの JAR パッケージを準備します。

ADB Spark ノードで使用する spark-examples_2.12-3.2.0.jar サンプル JAR パッケージをダウンロードします。
サンプルコードを OSS にアップロードします。
1. OSS コンソールにログインします。左側のナビゲーションウィンドウで、[バケット] をクリックします。
2. [バケット] ページで、[バケットの作成] をクリックします。[バケットの作成] パネルで、AnalyticDB for MySQL クラスターと同じリージョンにバケットを作成します。
  
  説明
  このトピックでは、dw-1127 という名前のバケットを例として使用します。
3. 外部ストレージディレクトリを作成します。
  
  作成が完了したら、[バケットに移動] をクリックします。ファイル一覧 ページで ディレクトリの作成 をクリックし、ディレクトリ名 を db_home に設定します。
4. サンプルコードファイル spark-examples_2.12-3.2.0.jar を db_home ディレクトリにアップロードします。詳細については、「オブジェクトのアップロード」をご参照ください。

ADB Spark ノードの設定

以下のパラメーターの説明に基づいて、ADB Spark ノードのコンテンツを設定します。

言語	パラメーター	説明
Java/Scala	[メイン Jar リソース]	JAR パッケージへの OSS パス。例：`oss://dw-1127/db_home/spark-examples_2.12-3.2.0.jar`
	[Main Class]	コンパイルされた JAR パッケージ内のメインクラス。サンプルコードのメインクラスの名前は `org.apache.spark.examples.SparkPi` です。
	[パラメータ]	コードに渡す引数。このフィールドは、`${var}` 形式の動的パラメーターで設定できます。説明この例では、動的パラメーター `${var}` を `1000` に設定できます。
	[設定項目]	Spark プログラムの実行時パラメーター。詳細については、「Spark アプリケーションの設定パラメーター」をご参照ください。例： `spark.driver.resourceSpec:medium`

Python

Python ファイルとデータの準備

次の手順に従って、テストデータファイルとサンプルコードを OSS にアップロードします。これにより、ノード設定のサンプルコードがテストデータファイルを読み取れるようになります。

テストデータを準備します。

data.txt という名前のファイルを作成し、次の内容を追加します。
```
Hello,Dataworks
Hello,OSS
```

サンプルコードを記述します。

spark_oss.py という名前のファイルを作成し、次の内容を spark_oss.py ファイルに追加します。

import sys

from pyspark.sql import SparkSession

# Spark を初期化します。
spark = SparkSession.builder.appName('OSS Example').getOrCreate()
# 指定されたファイルを読み取ります。ファイルパスは、ジョブに渡された引数の値によって指定されます。
textFile = spark.sparkContext.textFile(sys.argv[1])
# ファイルの行数を計算して出力します。
print("File total lines: " + str(textFile.count()))
# ファイルの最初の行を出力します。
print("First line is: " + textFile.first())

テストデータとサンプルコードを OSS にアップロードします。
1. OSS コンソールにログインします。左側のナビゲーションウィンドウで、[バケット] をクリックします。
2. [バケット] ページで、[バケットの作成] をクリックします。[バケットの作成] パネルで、AnalyticDB for MySQL クラスターと同じリージョンにバケットを作成します。
  
  説明
  このトピックでは、dw-1127 という名前のバケットを例として使用します。
3. 外部ストレージディレクトリを作成します。
  
  作成が完了したら、[バケットに移動] をクリックします。ファイル一覧 ページで ディレクトリの作成 をクリックし、ディレクトリ名 を db_home に設定します。
4. テストデータファイル data.txt とサンプルコードファイル spark_oss.py を db_home ディレクトリにアップロードします。詳細については、「オブジェクトのアップロード」をご参照ください。

ADB Spark ノードの設定

以下のパラメーターの説明に基づいて、ADB Spark ノードのコンテンツを設定します。

言語	パラメーター	説明
[Python]	[メインプログラムパッケージ]	サンプルコードファイルへのパス。例：`oss://dw-1127/db_home/spark_oss.py`
	[パラメータ]	コードに渡す引数。この例では、テストデータファイルのストレージパスを使用します。例：`oss://dw-1127/db_home/data.txt`
	[設定項目]	Spark プログラムの実行時パラメーター。詳細については、「Spark アプリケーションの設定パラメーター」をご参照ください。例： `spark.driver.resourceSpec:medium`

ステップ 2：ADB Spark ノードのデバッグ

ADB Spark ノードのデバッグプロパティを設定します。

ノードエディターの右側にある デバッグの構成 ペインで、計算リソース、ADB 計算リソースグループ、リソースグループ、および Compute Units パラメーターを設定します。

パラメータータイプ	パラメーター	説明
[計算リソース]	[計算リソース]	アタッチされている AnalyticDB for Spark コンピュートエンジンを選択します。
[計算リソース]	[ADB 計算リソースグループ]	AnalyticDB for MySQL クラスターで作成したジョブリソースグループを選択します。詳細については、「リソースグループの概要」をご参照ください。
[リソースグループ]	[リソースグループ]	コンピュートエンジンをアタッチしたときに接続性テストに合格したリソースグループを選択します。
[リソースグループ]	[Compute Units]	このノードはデフォルトの CU 数を使用します。この値を変更する必要はありません。

ADB Spark ノードをデバッグして実行します。

ノードタスクを実行するには、保存をクリックしてから実行をクリックします。

ステップ 3：ADB Spark ノードのスケジューリング

ADB Spark ノードのスケジューリングプロパティを設定します。

ノードタスクを定期的に実行するには、ノードエディターの右側にある スケジューリング設定 ペインを開きます。スケジューリングポリシー セクションで、必要に応じて次のパラメーターを設定します。他のパラメーターの詳細については、「ノードのスケジューリング設定」をご参照ください。

パラメーター	説明
[計算リソース]	アタッチされている AnalyticDB for Spark コンピュートエンジンを選択します。
[ADB 計算リソースグループ]	AnalyticDB for MySQL クラスターで作成したジョブリソースグループを選択します。詳細については、「リソースグループの概要」をご参照ください。
[リソースグループ]	コンピュートエンジンをアタッチしたときに接続性テストに合格したリソースグループを選択します。
[Compute Units]	このノードはデフォルトの CU 数を使用します。この値を変更する必要はありません。

ADB Spark ノードを公開します。

ノードタスクを設定した後、ノードを公開する必要があります。詳細については、「ノードまたはワークフローの公開」をご参照ください。

次のステップ

タスクを公開した後、オペレーションセンターで実行ステータスを表示できます。詳細については、「オペレーションセンター入門」をご参照ください。