Azure Data Factory と Synapse Analytics で Spark アクティビティを使用してデータを変換する

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータ サイエンス、リアルタイム分析、ビジネス インテリジェンス、レポートまで、あらゆるものをカバーしています。 無料で新しい試用版を開始する方法について説明します。

データ ファクトリと Synapse パイプラインの Spark アクティビティでは、独自のまたはオンデマンドの HDInsight クラスターで Spark プログラムを実行します。 この記事は、データ変換とサポートされる変換アクティビティの概要を説明する、 データ変換アクティビティ に関する記事に基づいています。 オンデマンドの Spark のリンクされたサービスを使用すると、サービスは自動的に Spark クラスターを作成し、Just-In-Time でデータを処理し、処理が完了するとクラスターを削除します。

UI を使用したパイプラインへの Spark アクティビティの追加

パイプラインに Spark アクティビティを使用するには、次の手順を実行します。

  1. パイプラインの [アクティビティ] ペイン内で Spark を検索し、Spark アクティビティをパイプライン キャンバスにドラッグします。

  2. まだ選択されていない場合は、キャンバスで新しい Spark アクティビティを選択します。

  3. HDI Cluster タブを選択して、Spark アクティビティの実行に使用する HDInsight クラスターへのリンク サービスを選択または新規作成します。

    Shows the UI for a Spark activity.

  4. Script / Jar タブを選択して、スクリプトをホストする Azure Storage アカウントにリンクされた新しい job サービスを選択または作成します。 そこで実行するファイルへのパスを指定します。 また、プロキシユーザー、デバッグ設定、スクリプトに渡す引数や Spark 設定パラメーターなど、詳細な設定を行うことができます。

    Shows the UI for the Script / Jar tab for a Spark activity.

Spark アクティビティのプロパティ

Spark アクティビティのサンプルの JSON 定義を次に示します。

{
    "name": "Spark Activity",
    "description": "Description",
    "type": "HDInsightSpark",
    "linkedServiceName": {
        "referenceName": "MyHDInsightLinkedService",
        "type": "LinkedServiceReference"
    },
    "typeProperties": {
        "sparkJobLinkedService": {
            "referenceName": "MyAzureStorageLinkedService",
            "type": "LinkedServiceReference"
        },
        "rootPath": "adfspark",
        "entryFilePath": "test.py",
        "sparkConfig": {
            "ConfigItem1": "Value"
        },
        "getDebugInfo": "Failure",
        "arguments": [
            "SampleHadoopJobArgument1"
        ]
    }
}

次の表で、JSON 定義で使用される JSON プロパティについて説明します。

プロパティ Description 必須
name パイプラインのアクティビティの名前。 はい
description アクティビティの動作を説明するテキスト。 いいえ
type Spark アクティビティの場合、アクティビティの種類は HDInsightSpark です。 はい
linkedServiceName Spark プログラムが実行されている HDInsight Spark のリンクされたサービスの名前。 このリンクされたサービスの詳細については、計算のリンクされたサービスに関する記事をご覧ください。 はい
SparkJobLinkedService Spark ジョブ ファイル、依存関係、およびログが含まれる Azure Storage のリンクされたサービス。 ここでは Azure Blob Storage および ADLS Gen2 にリンクされたサービスのみがサポートされています。 指定しない場合は、HDInsight クラスターに関連付けられているストレージが使用されます。 このプロパティの値には、Azure Storage のリンクされたサービスのみを指定できます。 いいえ
rootPath Azure BLOB コンテナーと Spark ファイルを含むフォルダー。 ファイル名は大文字と小文字が区別されます。 このフォルダーの構造の詳細については、「フォルダー構造」(次のセクション) をご覧ください。 はい
entryFilePath Spark コード/パッケージのルート フォルダーへの相対パス。 エントリ ファイルは、Python ファイルまたは .jar ファイルのいずれかにする必要があります。 はい
className アプリケーションの Java/Spark のメイン クラス いいえ
引数 Spark プログラムのコマンドライン引数の一覧です。 いいえ
proxyUser Spark プログラムの実行を偽装する借用すユーザー アカウント いいえ
sparkConfig Spark Configuration - Application properties (Spark 構成 - アプリケーションのプロパティ)」と題するトピックに示されている Spark 構成プロパティの値を指定します。 いいえ
getDebugInfo HDInsight クラスターで使用されている Azure Storage または sparkJobLinkedService で指定された Azure Storage に Spark ログ ファイルがコピーされるタイミングを指定します。 使用できる値は以下の通りです。None、Always、または Failure。 既定値:[なし] : いいえ

フォルダー構造

Spark ジョブは、Pig/Hive ジョブよりも拡張性に優れています。 Spark ジョブの場合、jar パッケージ (java CLASSPATH に配置)、Python ファイル (PYTHONPATH に配置) など、複数の依存関係を利用できます。

HDInsight のリンクされたサービスによって参照される Azure Blob Storage に、次のフォルダー構造を作成します。 その後、依存ファイルを、entryFilePath で表されるルート フォルダー内の適切なサブフォルダーにアップロードします。 たとえば、Python ファイルはルート フォルダーの pyFiles サブフォルダーに、jar ファイルはルート フォルダーの jar サブフォルダーにアップロードします。 実行時にサービスに必要な Azure Blob Storage のフォルダー構造を次に示します。

パス 説明 必須 Type
. (ルート) ストレージのリンクされたサービスにおける Spark ジョブのルート パス はい Folder
<user defined > Spark ジョブの入力ファイルを指定するパス はい ファイル
./jars このフォルダーのすべてのファイルがアップロードされ、クラスターの Java classpath に配置されます。 いいえ Folder
./pyFiles このフォルダーのすべてのファイルがアップロードされ、クラスターの PYTHONPATH に配置されます いいえ Folder
./files このフォルダーのすべてのファイルがアップロードされ、Executor 作業ディレクトリに配置されます いいえ Folder
./archives このフォルダーのファイルは圧縮されていません いいえ Folder
./logs Spark クラスターのログが格納されているフォルダー。 いいえ Folder

次の例のストレージには、HDInsight のリンクされたサービスによって参照される Azure Blob Storage に 2 つの Spark ジョブ ファイルが含まれています。

SparkJob1
	main.jar
	files
		input1.txt
		input2.txt
	jars
		package1.jar
		package2.jar
	logs
	
	archives
	
	pyFiles

SparkJob2
	main.py
	pyFiles
		scrip1.py
		script2.py
	logs
	
	archives
	
	jars
	
	files
	

別の手段でデータを変換する方法を説明している次の記事を参照してください。