Apache Spark を使用した対話型データ整形

データラングリングは、機械学習プロジェクトの重要な側面です。この記事では、Azure Synapse によってサポートされるサーバーレス Apache Spark コンピューティングで Azure Machine Learning ノートブックを実行して、対話型のデータラングリングを実行する方法について説明します。

この記事では、サーバーレス Spark コンピューティングをアタッチして構成する方法について説明します。次に、サーバーレス Spark コンピューティングを使用して、複数のソースからデータにアクセスしてラングリングする方法について説明します。

前提条件

Azure サブスクリプションの所有者またはロールの割り当て権限。無料の Azure アカウントを作成できます。
Azure Machine Learning ワークスペース。詳細については、ワークスペースリソースの作成に関するページを参照してください。
ワークスペース内の既定のファイル共有にアップロードされた titanic.csv データセット。
Azure Data Lake Storage Gen 2 ストレージアカウント。詳細については、「 Azure Data Lake Storage Gen 2 ストレージアカウントの作成」を参照してください。
次のロールの割り当てが付与されます。
- Azure Storage アカウントにアクセスするには、Contributor ロールと Storage Blob Data Contributor ロールが Azure Storage アカウントに必要です。
- Azure Key Vault のシークレットにアクセスするには、キーコンテナーシークレットユーザー ロールがキーコンテナーに必要です。

詳細については、以下を参照してください。

ノートブックセッションでサーバーレス Spark コンピューティングを使用する

サーバーレス Spark コンピューティングを使用することは、対話型データラングリングのために Spark クラスターにアクセスする最も簡単な方法です。 Synapse Spark プールに接続されたフルマネージドのサーバーレス Spark コンピューティングは、Azure Machine Learning ノートブックで直接使用できます。

次のいずれかのデータアクセスとラングリングソースとメソッドを使用するには、ファイルまたはノートブックページの上部にある [コンピューティング] の横にある Azure Machine Learning サーバーレス Spark>Serverless Spark Compute - Available を選択して Spark サーバーレス コンピューティング をアタッチします。コンピューティングがセッションにアタッチされるまでに 1 ~ 2 分かかる場合があります。

サーバーレス Spark セッションを構成する

サーバーレス Spark コンピューティングをアタッチした後、いくつかの値を設定または変更することで、Spark セッションを構成できます。 Spark セッションを構成するには:

ファイルまたはノートブックページの左上にある [ セッションの構成] を 選択します。
[ セッションの構成 ] 画面で、次のいずれかの設定を変更します。
- [コンピューティング] ウィンドウで、次の操作を行います。
  - [ノードサイズ] のドロップダウンメニューから別のサイズを選択して、マシン のサイズを変更します。
  - Executor を動的に割り当てるかどうかを選択します。
  - Spark セッションの Executor の数を選択します。
  - ドロップダウンメニューから使用可能な場合は、別の Executor サイズ を選択します。
- [設定] ウィンドウで、次の操作を行います。
  - Apache Spark のバージョンを 3.5 以外のバージョン (使用可能な場合) に変更します。
    
    Important
    
    Azure Synapse Runtime for Apache Spark 3.4 は、2026 年 3 月 31 日にサポートが終了します。継続的なサポートのために Apache Spark 3.5 に移行します。詳細については、 Azure Synapse ランタイムに関するページを参照してください。
  - セッションタイムアウトを防ぐのに役立つセッションタイムアウト値を分単位で大きい数値に変更します。
  - [ 構成設定] で、 プロパティ 名/値の設定を追加して、必要に応じてセッションを構成します。
    
    ヒント
    
    セッションレベルの Conda パッケージを使用する場合は、値が spark.hadoop.aml.enable_cache の true 構成プロパティを追加すると、Spark セッションのコールド開始時刻が短縮される可能性があります。セッションレベルの Conda パッケージを使用したセッションコールドスタートは、通常、最初に 10 分から 15 分かかります。以降のセッションコールドは、構成変数が true に設定された状態で開始されます。通常、3 ~ 5 分かかります。
- [Python パッケージ] ウィンドウで、次の操作を行います。
  - Conda ファイルを使用してセッションを構成するには、[ Conda ファイルのアップロード] を選択します。 [conda ファイルの選択] の横にある [参照] を選択し、コンピューター上の適切な Conda YAML ファイルを参照して開いてアップロードします。
  - カスタム環境を使用するには、[ カスタム環境 ] を選択し、[環境の種類] でカスタム環境を選択します。詳細については、「ソフトウェア環境の管理」を参照してください。
[ 適用] を選択して、すべての構成を適用します。

セッション構成の変更は保持され、接続されているサーバーレス Spark コンピューティングを使用する他のノートブックセッションで使用できます。

Azure Data Lake Storage からのデータのインポートとラングリング

Azure Data Lake Storage アカウントに格納されているデータにアクセスしてラングリングするには、abfss://またはサービスプリンシパルベースのアクセス権を持つプロトコル URI を使用します。ユーザー ID パススルーには、追加の構成は必要ありません。

いずれかの方法を使用するには、ユーザー ID またはサービスプリンシパルに、Azure Data Lake Storage アカウントの 共同作成者 ロールと ストレージ BLOB データ共同作成者ロールの割り当てが必要です。

ユーザー ID パススルーの場合は、次のデータラングリングコードサンプルを実行して、abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/<PATH_TO_DATA>でpyspark.pandas形式のデータ URI を使用します。 <STORAGE_ACCOUNT_NAME> プレースホルダーを Azure Data Lake Storage アカウントの名前に置き換え、<FILE_SYSTEM_NAME>をデータコンテナーの名前に置き換えます。

import pyspark.pandas as pd

df = pd.read_csv(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv",
    index_col="PassengerId",
)
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(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled",
    index_col="PassengerId",
)

サービスプリンシパルを使用する

サービスプリンシパルを使用して Azure Data Lake Storage からデータにアクセスしてラングリングするには、最初に次のようにサービスプリンシパルを設定します。

サービスプリンシパルを作成し、必要なストレージ BLOB データ共同作成者と Key Vault シークレットのユーザーロールを割り当てます。
アプリの登録からサービスプリンシパルテナント ID、クライアント ID、クライアントシークレットの値を取得し、値の Azure Key Vault シークレットを作成します。

セッション構成に次のプロパティ名と値のペアを追加して、サービスプリンシパルのテナント ID、クライアント ID、およびクライアントシークレットを設定します。 <STORAGE_ACCOUNT_NAME>をストレージアカウント名に置き換え、<TENANT_ID>をサービスプリンシパルテナント ID に置き換えます。

プロパティ名	価値
`fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`	アプリケーション (クライアント) ID 値
`fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`	`https://login.microsoftonline.com/<TENANT_ID>/oauth2/token`
`fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`	クライアントシークレットの値

次のコードを実行します。コード内の get_secret() 呼び出しは、Key Vault 名と、サービスプリンシパルテナント ID、クライアント ID、クライアントシークレット用に作成された Key Vault シークレットの名前によって異なります。

from pyspark.sql import SparkSession

sc = SparkSession.builder.getOrCreate()
token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary

# Set up service principal tenant ID, client ID, and secret from Azure Key Vault
client_id = token_library.getSecret("<KEY_VAULT_NAME>", "<CLIENT_ID_SECRET_NAME>")
tenant_id = token_library.getSecret("<KEY_VAULT_NAME>", "<TENANT_ID_SECRET_NAME>")
client_secret = token_library.getSecret("<KEY_VAULT_NAME>", "<CLIENT_SECRET_NAME>")

# Set up a service principal that has access to the data
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.auth.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net", "OAuth"
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth.provider.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    client_id,
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    client_secret,
)
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net",
    "https://login.microsoftonline.com/" + tenant_id + "/oauth2/token",
)

コードサンプルに示すように、形式のデータ URI を使用して、titanic.csvデータをインポートしてラングリングします。 <STORAGE_ACCOUNT_NAME> プレースホルダーを Azure Data Lake Storage アカウントの名前に置き換え、<FILE_SYSTEM_NAME>をデータコンテナーの名前に置き換えます。

import pyspark.pandas as pd

df = pd.read_csv(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv",
    index_col="PassengerId",
)
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(
    "abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled",
    index_col="PassengerId",
)

Azure BLOB ストレージからのデータのインポートとラングリング

Azure Blob Storage データには、 ストレージアカウントアクセスキー または Shared Access Signature (SAS) トークンのいずれかを使用してアクセスできます。資格情報をシークレットとして Azure Key Vault に格納し、Spark セッション構成でプロパティとして設定します。

次のいずれかのコードスニペットを実行します。コードスニペットの get_secret() 呼び出しには、キーコンテナーの名前と、Azure Blob Storage アカウントアクセスキーまたは SAS トークン用に作成されたシークレットの名前が必要です。
- ストレージアカウントのアクセスキーを構成するには、次のコードスニペットに示すように、 fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net プロパティを設定します。
```
from pyspark.sql import SparkSession

sc = SparkSession.builder.getOrCreate()
token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
access_key = token_library.getSecret("<KEY_VAULT_NAME>", "<ACCESS_KEY_SECRET_NAME>")
sc._jsc.hadoopConfiguration().set(
    "fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net", access_key
)
```
- SAS トークンを構成するには、次のコードスニペットに示すように、 fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net プロパティを設定します。
```
from pyspark.sql import SparkSession

sc = SparkSession.builder.getOrCreate()
token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
sas_token = token_library.getSecret("<KEY_VAULT_NAME>", "<SAS_TOKEN_SECRET_NAME>")
sc._jsc.hadoopConfiguration().set(
    "fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net",
    sas_token,
)
```

wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/<PATH_TO_DATA>形式のデータ URI を使用して、次のデータラングリングコードを実行します。

import pyspark.pandas as pd

df = pd.read_csv(
    "wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/titanic.csv",
    index_col="PassengerId",
)
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(
    "wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/wrangled",
    index_col="PassengerId",
)

Azure Machine Learning データストアからのデータのインポートとラングリング

Azure Machine Learning データストアからデータにアクセスするには、URI 形式でデータストア上のデータへのパスを定義しますazureml://datastores/<DATASTORE_NAME>/paths/<PATH_TO_DATA>。

次のコードサンプルを実行し、titanic.csvデータをAzure Machine Learning データストアから読み取り、azureml://データストアURI とpyspark.pandasを使って整形します。

import pyspark.pandas as pd

df = pd.read_csv(
    "azureml://datastores/<DATASTORE_NAME>/paths/data/titanic.csv",
    index_col="PassengerId",
)
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(
    "azureml://datastores/<DATASTORE_NAME>/paths/data/wrangled",
    index_col="PassengerId",
)

Azure Machine Learning データストアは、Azure ストレージアカウントのアクセスキー、SAS トークン、サービスプリンシパルの資格情報、または資格情報のないデータアクセスを使用してデータにアクセスできます。データストアの種類と基になる Azure ストレージアカウントの種類に応じて、適切な認証メカニズムを選択します。

次の表は、Azure Machine Learning データストア内のデータにアクセスするための認証メカニズムをまとめたものです。

ストレージアカウントの種類	資格情報のないデータアクセス	データアクセスメカニズム	ロールの割り当て
Azure BLOB	いいえ	アクセスキーまたは SAS トークン	ロールの割り当ては必要ありません。
Azure BLOB	はい	ユーザー ID パススルー*	ユーザー ID には、Azure Blob Storage アカウントで適切なロールの割り当てが必要です。
Azure Data Lake Storage	いいえ	サービスプリンシパル	サービスプリンシパルには、Azure Data Lake Storage ストレージアカウントに適切なロールの割り当てが必要です。
Azure Data Lake Storage	はい	ユーザー ID パススルー	ユーザー ID には、Azure Data Lake Storage ストレージアカウントで適切なロールの割り当てが必要です。

* ユーザー ID パススルーは、論理的な削除が有効になっていない場合にのみ、Azure Blob Storage アカウントを指す資格情報のないデータストアに対して機能します。

Azure Machine Learning Studio の既定のワークスペースファイル共有は、Notebooks の [ファイル] タブの下のディレクトリツリーです。ノートブックコードは、他の構成なしでファイルの絶対パスを使用して、 file:// プロトコルを使用して、このファイル共有に格納されているファイルに直接アクセスできます。既定のファイル共有は、サーバーレス Spark コンピューティングプールとアタッチされた Synapse Spark プールの両方にマウントされます。

次のコードスニペットは、既定のファイル共有のユーザー名のすぐ下にあるデータフォルダーに格納されている titanic.csv ファイルから データ にアクセスし、ラングリングします。 <USER> プレースホルダーをユーザー名に置き換えます。

import os
import pyspark.pandas as pd

abspath = os.path.abspath(".")
file = "file://" + abspath + "/Users/<USER>/data/titanic.csv"
print(file)
df = pd.read_csv(file, index_col="PassengerId")
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
output_path = "file://" + abspath + "/Users/<USER>/data/wrangled"
df.to_csv(output_path, index_col="PassengerId")

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-03-11

次の方法で共有

Apache Spark を使用した対話型データ整形

前提条件

ノートブック セッションでサーバーレス Spark コンピューティングを使用する

サーバーレス Spark セッションを構成する

Azure Data Lake Storage からのデータのインポートとラングリング

サービス プリンシパルを使用する

Azure BLOB ストレージからのデータのインポートとラングリング

Azure Machine Learning データストアからのデータのインポートとラングリング

既定のファイル共有のデータにアクセスする

関連コンテンツ

フィードバック

その他のリソース

ノートブックセッションでサーバーレス Spark コンピューティングを使用する

サービスプリンシパルを使用する