Azure Machine Learning の Apache Spark ジョブの構成

2024-09-04

適用対象:Azure CLI ml extension v2 (現行)Python SDK azure-ai-ml v2 (現行)

Azure Machine Learning 統合と Azure Synapse Analytics を併用すると、Azure Synapse を利用する分散コンピューティング機能に簡単にアクセスできます。これにより、Azure Machine Learning 上の Apache Spark ジョブをスケーリングすることができます。

この記事では、Azure Machine Learning サーバーレス Spark コンピューティング、Azure Data Lake Storage (ADLS) Gen 2 ストレージアカウント、ユーザー ID パススルーを使って Spark ジョブをいくつかの簡単な手順で送信する方法について説明します。

Azure Machine Learning での Apache Spark の概念の詳細については、こちらのリソースを参照してください。

前提条件

適用対象:Azure CLI ml 拡張機能 v2 (現行)

Azure サブスクリプション。Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。
Azure Machine Learning ワークスペース。詳細については、ワークスペースリソースの作成に関するページを参照してください。
Azure Data Lake Storage (ADLS) Gen 2 ストレージアカウント。詳細については、Azure Data Lake Storage (ADLS) Gen 2 ストレージアカウントの作成に関するページを参照してください。
Azure Machine Learning コンピューティングインスタンスを作成します。
Azure Machine Learning CLI をインストールする。

Azure ストレージアカウントにロールの割り当てを追加する

Apache Spark ジョブを送信する前に、入力と出力のデータパスにアクセスできることを確認する必要があります。ログインユーザーのユーザー ID に共同作成者と ストレージ BLOB データ共同作成者のロールを割り当て、読み取りと書き込みのアクセスを有効にします。

ユーザー ID に適切なロールを割り当てるには:

Microsoft Azure portalを開きます。
ストレージアカウント サービスを検索して選択します。
[ストレージアカウント] ページで、一覧から Azure Data Lake Storage (ADLS) Gen 2 ストレージアカウントを選択します。ストレージアカウントの概要を示すページが開きます。
左ペインから [アクセス制御 (IAM)] を選択します。
[ロールの割り当ての追加] を選択します。
ストレージ BLOB データ共同作成者ロールを検索します。
[ストレージ BLOB データ共同作成者] ロールを選択します。
[次へ] を選択します。
ユーザー、グループ、またはサービスプリンシパル を選択します。
[+ メンバーの選択] を選択します。
[選択] の下のテキストボックスで、ユーザー ID を検索します。
[選択したメンバー] の下に表示されるように、リストからユーザー ID を選びます。
適切なユーザー ID を選びます。
[次へ] を選択します。
[レビューと割り当て] を選択します。
共同作成者ロールの割り当てについて手順 2 から 13 を繰り返します。

ユーザー ID に適切なロールが割り当てられると、Azure Data Lake Storage (ADLS) Gen 2 ストレージアカウント内のデータにアクセスできるようになります。

パラメーター化された Python コードを作成する

Spark ジョブには、引数を受け取る Python スクリプトが必要です。このスクリプトをビルドするために、対話型データラングリングから開発された Python コードを変更できます。サンプルの Python スクリプトを次に示します。

# titanic.py
import argparse
from operator import add
import pyspark.pandas as pd
from pyspark.ml.feature import Imputer

parser = argparse.ArgumentParser()
parser.add_argument("--titanic_data")
parser.add_argument("--wrangled_data")

args = parser.parse_args()
print(args.wrangled_data)
print(args.titanic_data)

df = pd.read_csv(args.titanic_data, index_col="PassengerId")
imputer = Imputer(inputCols=["Age"], outputCol="Age").setStrategy(
    "mean"
)  # Replace missing values in Age column with the mean value
df.fillna(
    value={"Cabin": "None"}, inplace=True
)  # Fill Cabin column with value "None" if missing
df.dropna(inplace=True)  # Drop the rows which still have any missing value
df.to_csv(args.wrangled_data, index_col="PassengerId")

注

この Python コードサンプルでは、pyspark.pandas を使用します。これをサポートしているのは Spark ランタイムバージョン 3.2 のみです。
titanic.py ファイルが src という名前のフォルダーにアップロードされていることを確認してください。 src フォルダーは、Python スクリプト/ノートブック、またはスタンドアロン Spark ジョブを定義する YAML 仕様ファイルを作成したのと同じディレクトリにある必要があります。

そのスクリプトには --titanic_data と --wrangled_data の次の 2 つの引数があります。これらの引数は、それぞれ入力データパスと出力フォルダーを渡します。スクリプトでは titanic.csv ファイルを使用します (こちらで入手可能）。このファイルを、Azure Data Lake Storage (ADLS) Gen 2 ストレージアカウントで作成されたコンテナーにアップロードします。

スタンドアロン Spark ジョブを送信する

適用対象:Azure CLI ml 拡張機能 v2 (現行)

ヒント

Spark ジョブは以下から送信できます。

Azure Machine Learning コンピューティングインスタンスのターミナル。
Azure Machine Learning コンピューティングインスタンスに接続された Visual Studio Code のターミナル。
Azure Machine Learning CLI がインストールされているローカルコンピューター。

この YAML 仕様の例では、スタンドアロンの Spark ジョブを示します。 Azure Machine Learning サーバーレス Spark コンピューティング、ユーザー ID パススルー、abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> という形式の入力および出力データの URI が使われます。ここでは、<FILE_SYSTEM_NAME> はコンテナー名と一致します。

$schema: http://azureml/sdk-2-0/SparkJob.json
type: spark

code: ./src 
entry:
  file: titanic.py

conf:
  spark.driver.cores: 1
  spark.driver.memory: 2g
  spark.executor.cores: 2
  spark.executor.memory: 2g
  spark.executor.instances: 2

inputs:
  titanic_data:
    type: uri_file
    path: abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv
    mode: direct

outputs:
  wrangled_data:
    type: uri_folder
    path: abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled/
    mode: direct

args: >-
  --titanic_data ${{inputs.titanic_data}}
  --wrangled_data ${{outputs.wrangled_data}}

identity:
  type: user_identity

resources:
  instance_type: standard_e4s_v3
  runtime_version: "3.2"

上の YAML 仕様ファイルの内容:

code プロパティには、パラメーター化された titanic.py ファイルを含むフォルダーの相対パスを定義します。
resource プロパティは、サーバーレス Spark コンピューティングで使用される instance_type と Apache Spark runtime_version の値を定義します。現在サポートされているインスタンスの種類の値は次のとおりです。
- standard_e4s_v3
- standard_e8s_v3
- standard_e16s_v3
- standard_e32s_v3
- standard_e64s_v3

この YAML ファイルを az ml job create コマンドの --file パラメーターで指定して、次のようにスタンドアロン Spark ジョブを作成できます。

az ml job create --file <YAML_SPECIFICATION_FILE_NAME>.yaml --subscription <SUBSCRIPTION_ID> --resource-group <RESOURCE_GROUP> --workspace-name <AML_WORKSPACE_NAME>

適用対象: Python SDK azure-ai-ml v2 (現行)

ヒント

Spark ジョブは以下から送信できます。

Azure Machine Learning コンピューティングインスタンスに接続された Azure Machine Learning ノートブック。
Azure Machine Learning コンピューティングインスタンスに接続された Visual Studio Code。
Azure Machine Learning SDK for Python がインストールされているローカルコンピューター。

この Python コードスニペットは、スタンドアロンの Spark ジョブの作成を示しています。 Azure Machine Learning サーバーレス Spark コンピューティング、ユーザー ID パススルー、abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> という形式の入力および出力データの URI が使われます。ここでは、<FILE_SYSTEM_NAME> はコンテナー名と一致します。

from azure.ai.ml import MLClient, spark, Input, Output
from azure.identity import DefaultAzureCredential
from azure.ai.ml.entities import UserIdentityConfiguration

subscription_id = "<SUBSCRIPTION_ID>"
resource_group = "<RESOURCE_GROUP>"
workspace_name = "<AML_WORKSPACE_NAME>"
ml_client = MLClient(
    DefaultAzureCredential(), subscription_id, resource_group, workspace_name
)

spark_job = spark(
    display_name="Titanic-Spark-Job-SDK",
    code="./src",
    entry={"file": "titanic.py"},
    driver_cores=1,
    driver_memory="2g",
    executor_cores=2,
    executor_memory="2g",
    executor_instances=2,
    resources={
        "instance_type": "Standard_E8S_V3",
        "runtime_version": "3.2.0",
    },
    inputs={
        "titanic_data": Input(
            type="uri_file",
            path="abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv",
            mode="direct",
        ),
    },
    outputs={
        "wrangled_data": Output(
            type="uri_folder",
            path="abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled/",
            mode="direct",
        ),
    },
    identity=UserIdentityConfiguration(),
    args="--titanic_data ${{inputs.titanic_data}} --wrangled_data ${{outputs.wrangled_data}}",
)

returned_spark_job = ml_client.jobs.create_or_update(spark_job)

# Wait until the job completes
ml_client.jobs.stream(returned_spark_job.name)

上のコードサンプルの内容:

code パラメーターには、パラメーター化された titanic.py ファイルを含むフォルダーの相対パスを定義します。
サーバーレス Spark コンピューティング (プレビュー) で使用される resource と Apache Spark instance_type を定義する runtime_version パラメーター。現在サポートされているインスタンスの種類の値は次のとおりです。
- Standard_E4S_V3
- Standard_E8S_V3
- Standard_E16S_V3
- Standard_E32S_V3
- Standard_E64S_V3

重要

現在、この機能はパブリックプレビュー段階にあります。このプレビューバージョンはサービスレベルアグリーメントなしで提供されており、運用環境のワークロードに使用することは推奨されません。特定の機能はサポート対象ではなく、機能が制限されることがあります。

詳しくは、Microsoft Azure プレビューの追加使用条件に関するページをご覧ください。

まず、パラメーター化された Python コード titanic.py を、ワークスペースの既定のデータストア workspaceblobstore 用の Azure Blob Storage コンテナーにアップロードします。 Azure Machine Learning スタジオの UI を使ってスタンドアロン Spark ジョブを送信するには:

画面の右上付近にある [+ 新規] を選択します。
[Spark ジョブ (プレビュー)] を選択びます。
[コンピューティング] 画面で:
1. [コンピューティングの種類を選択] で、サーバーレス Spark コンピューティングに [Spark サーバーレス] を選択します。
2. [仮想マシンサイズ] を選びます。現在サポートされているインスタンスの種類は次のとおりです。
  - Standard_E4s_v3
  - Standard_E8s_v3
  - Standard_E16s_v3
  - Standard_E32s_v3
  - Standard_E64s_v3
3. [Spark ランタイムのバージョン] に [Spark 3.2] を選びます。
4. [次へ] を選択します。
[環境] 画面で [次へ] を選びます。
[ジョブの設定] 画面で、次のようにします。
1. ジョブの [名前] を指定するか、既定で生成されたジョブの [名前] を使います。
2. ドロップダウンメニューから [実験名] を選びます。
3. [タグの追加] で、[名前] と [値] を指定してから、[追加] を選びます。タグの追加は省略できます。
4. [コード] セクションで:
  1. [Choose code ___location] (コードの場所の選択) ドロップダウンから [Azure Machine Learning workspace default blob storage] (Azure Machine Learning ワークスペースの既定の BLOB ストレージ) を選びます。
  2. [アップロードするコードファイルへのパス] で、[参照] を選びます。
  3. [パスの選択] というタイトルのポップアップ画面で、ワークスペースの既定のデータストア titanic.py 上のコードファイル workspaceblobstore のパスを選びます。
  4. [保存] を選択します。
  5. スタンドアロンジョブの titanic.py の名前として「」を入力します。
  6. 入力を追加するには、[入力] で [+ 入力の追加] を選んで、次のようにします
    1. [入力名] に「titanic_data」と入力します。入力は、後の手順の [引数] でこの名前を参照する必要があります。
    2. [入力の種類] に [データ] を選びます。
    3. [データの種類] に [ファイル] を選びます。
    4. [データソース] に [URI] を選びます。
    5. titanic.csv ファイルの Azure Data Lake Storage (ADLS) Gen 2 データの URI を abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> という形式で入力します。ここでは、<FILE_SYSTEM_NAME> はコンテナー名と一致します。
  7. 入力を追加するには、[出力] で [+ 出力の追加] を選んで、次のようにします
    1. [出力名] に「wrangled_data」と入力します。出力は、後の手順の [引数] でこの名前を参照する必要があります。
    2. [出力の種類] に [フォルダー] を選びます。
    3. [Output URI destination] (出力 URI の宛先) に Azure Data Lake Storage (ADLS) Gen 2 フォルダーの URI を abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA> という形式で入力します。ここでは、<FILE_SYSTEM_NAME> はコンテナー名と一致します。
  8. [引数] に「--titanic_data ${{inputs.titanic_data}} --wrangled_data ${{outputs.wrangled_data}}」と入力します。
5. [Spark の構成] セクションで:
  1. [Executor のサイズ] で:
    1. Executor の [コア] の数に「2」、Executor の [メモリ (GB)] に「2」と入力します。
    2. [Dynamically allocated executors] (動的に割り当てられた Executor) に [無効] を選びます。
    3. [Executor instances] (Executor インスタンス) の数に「2」と入力します。
  2. [Driver size] (ドライバーサイズ) のドライバーの [コア] の数に「1」、[メモリ (GB)] に「2」と入力します。
6. [次へ] を選択します。
[レビュー] 画面で:
1. 送信する前にジョブの仕様を確認します。
2. [作成] を選んで、スタンドアロンの Spark ジョブを送信します。

注

Azure Machine Learning サーバーレス Spark コンピューティングを使ってスタジオ UI から送信されたスタンドアロンジョブは、データアクセスについてはユーザー ID パススルーが既定になります。

ヒント

Azure Synapse ワークスペースに既存の Synapse Spark プールが存在する場合があります。既存の Synapse Spark プールを使う場合は、Azure Machine Learning ワークスペース内の Synapse Spark プールをアタッチする手順に従ってください。

次の方法で共有

Azure Machine Learning の Apache Spark ジョブの構成

前提条件

Azure ストレージ アカウントにロールの割り当てを追加する

パラメーター化された Python コードを作成する

スタンドアロン Spark ジョブを送信する

次のステップ

フィードバック

その他のリソース

Azure ストレージアカウントにロールの割り当てを追加する