Azure Data Factory で Spark アクティビティを使用してクラウドのデータを変換する

[アーティクル]
10/03/2024

適用対象: Azure Data Factory Azure Synapse Analytics

ヒント

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

このチュートリアルでは、Azure Portal を使用して Azure Data Factory パイプラインを作成します。このパイプラインは、Spark アクティビティとオンデマンドの Azure HDInsight のリンクされたサービスを使用して、データを変換します。

このチュートリアルでは、以下の手順を実行します。

データファクトリを作成します。
Spark アクティビティを使用するパイプラインを作成します。
パイプラインの実行をトリガーする。
パイプラインの実行を監視します。

Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。

前提条件

Note

Azure を操作するには、Azure Az PowerShell モジュールを使用することをお勧めします。作業を始めるには、「Azure PowerShell をインストールする」を参照してください。 Az PowerShell モジュールに移行する方法については、「AzureRM から Az への Azure PowerShell の移行」を参照してください。

Azure ストレージアカウント。 Python スクリプトと入力ファイルを作成し、Azure Storage にアップロードします。 Spark プログラムからの出力は、このストレージアカウントに格納されます。オンデマンドの Spark クラスターでは、同じストレージアカウントがプライマリストレージとして使用されます。

Note

HDInsight は、Standard レベルで、汎用ストレージアカウントのみをサポートしています。アカウントが Premium または BLOB のみのストレージアカウントでないことを確認してください。

Azure PowerShell。 Azure PowerShell のインストールと構成の方法に関するページに記載されている手順に従います。

Python スクリプトを BLOB ストレージアカウントにアップロードする

次の内容が含まれた、WordCount_Spark.py という名前の Python ファイルを作成します。

import sys
from operator import add

from pyspark.sql import SparkSession

def main():
    spark = SparkSession\
        .builder\
        .appName("PythonWordCount")\
        .getOrCreate()

    lines = spark.read.text("wasbs://adftutorial@<storageaccountname>.blob.core.windows.net/spark/inputfiles/minecraftstory.txt").rdd.map(lambda r: r[0])
    counts = lines.flatMap(lambda x: x.split(' ')) \
        .map(lambda x: (x, 1)) \
        .reduceByKey(add)
    counts.saveAsTextFile("wasbs://adftutorial@<storageaccountname>.blob.core.windows.net/spark/outputfiles/wordcount")

    spark.stop()

if __name__ == "__main__":
    main()

<storageAccountName> を Azure ストレージアカウントの名前に置き換えます。その後、ファイルを保存します。
Azure BLOB ストレージで、adftutorial という名前のコンテナーを作成します (存在しない場合)。
spark という名前のフォルダーを作成します。
spark フォルダーの下に、script という名前のサブフォルダーを作成します。
WordCount_Spark.py ファイルを script サブフォルダーにアップロードします。

入力ファイルをアップロードする

minecraftstory.txt という名前のファイルを作成し、任意のテキストを入力しておきます。このテキストの単語数が Spark プログラムによってカウントされます。
spark フォルダーの下に、inputfiles という名前のサブフォルダーを作成します。
inputfiles サブフォルダーに minecraftstory.txt ファイルをアップロードします。

Data Factory の作成

使うデータファクトリがまだない場合は、記事「クイックスタート: Azure portal を使用してデータファクトリを作成する」の手順に従って作成します。

リンクされたサービスを作成します

このセクションでは、2 つのリンクされたサービスを作成します。

Azure ストレージアカウントをデータファクトリにリンクする、Azure Storage のリンクされたサービス。このストレージは、オンデマンドの HDInsight クラスターによって使用されます。ここには、実行される Spark スクリプトも含まれています。
オンデマンドの HDInsight のリンクされたサービス。 Azure Data Factory によって自動的に HDInsight クラスターが作成され、Spark プログラムが実行されます。 HDInsight クラスターは、事前に構成された時間だけアイドル状態になったら削除されます。

Azure Storage のリンクされたサービスを作成する

ホームページの左側のパネルで [管理] タブに切り替えます。
ウィンドウの下部にある [接続] を選択して、 [+ 新規] を選択します。
[New Linked Service](新しいリンクされたサービス) ウィンドウで、[データストア]>[Azure Blob Storage] を選択し、[続行] を選択します。
[ストレージアカウント名] で一覧から名前を選択し、[保存] を選択します。

オンデマンドの HDInsight のリンクされたサービスを作成する

[+ 新規] ボタンをもう一度選択して、別のリンクされたサービスを作成します。
[New Linked Service](新しいリンクされたサービス) ウィンドウで、[コンピューティング]>[Azure HDInsight] の順に選択し、[続行] を選択します。
[New Linked Service](新しいリンクされたサービス) ウィンドウで、次の手順を完了します。

a. [名前] に「AzureHDInsightLinkedService」と入力します。

b. [Type](タイプ) で [On-demand HDInsight](オンデマンド HDInsight) が選択されていることを確認します。

c. [Azure Storage Linked Service](Azure Storage のリンクされたサービス) で [AzureBlobStorage1] を選択します。このリンクされたサービスは、前の手順で作成したものです。別の名前を使用した場合は、ここで適切な名前を指定します。

d. [クラスターの種類] で [spark] を選択します。

e. [サービスプリンシパル ID] に、HDInsight クラスターを作成するアクセス許可を備えたサービスプリンシパルの ID を入力します。

このサービスプリンシパルは、サブスクリプションまたはクラスターが作成されるリソースグループの共同作成者ロールのメンバーである必要があります。詳細については、「Microsoft Entra アプリケーションとサービスプリンシパルの作成」を参照してください。 [サービスプリンシパル ID] は "アプリケーション ID" に、 [サービスプリンシパルキー] は "クライアントシークレット" の値に相当します。

f. [サービスプリンシパルキー] に、キーを入力します。

g. [リソースグループ] に、データファクトリの作成時に使用したのと同じリソースグループを選択します。 Spark クラスターは、このリソースグループに作成されます。

h. [OS の種類] を展開します。

i. [クラスターユーザー名] に名前を入力します。

j. そのユーザーのクラスターパスワードを入力します。

k. [完了] を選択します。

Note

Azure HDInsight には、サポートされる各 Azure リージョンで使用できるコアの合計数に制限があります。オンデマンドの HDInsight のリンクされたサービスの場合、HDInsight クラスターは、プライマリストレージとして使用されるのと同じ Azure ストレージの場所に作成されます。クラスターを正常に作成するための十分なコアクォータがあることを確認します。詳細については、「Hadoop、Spark、Kafka などの HDInsight クラスターをセットアップする」を参照してください。

パイプラインを作成する

+ (正符号) ボタンを選択し、メニューの [パイプライン] を選択します。
[アクティビティ] ツールボックスで [HDInsight] を展開します。 [アクティビティ] ツールボックスからパイプラインデザイナー画面に Spark アクティビティをドラッグします。
下部の Spark アクティビティウィンドウのプロパティで、次の手順を完了します。

a. [HDI Cluster](HDI クラスター) タブに切り替えます。

b. (前の手順で作成した) AzureHDInsightLinkedService を選択します。
[Script/Jar](スクリプト/Jar) タブに切り替えて、次の手順を実行します。

a. [Job Linked Service](ジョブのリンクされたサービス) で [AzureBlobStorage1] を選択します。

b. [ストレージを参照] を選択します。

c. adftutorial/spark/script フォルダーに移動します。WordCount_Spark.py を選択し、[完了] を選択します。
パイプラインを検証するために、ツールバーの [検証] ボタンを選択します。 >> (右矢印) ボタンを選択して、検証ウィンドウを閉じます。
[すべて公開] を選択します。 Data Factory UI により、エンティティ (リンクされたサービスとパイプライン) が Azure Data Factory サービスに公開されます。

パイプラインの実行をトリガーする

ツールバーの [トリガーの追加] を選択し、 [Trigger Now](今すぐトリガー) を選択します。

パイプラインの実行を監視します

[監視] タブに切り替えます。パイプラインが実行されていることを確認します。 Spark クラスターの作成には約 20 分かかります。
[最新の情報に更新] を定期的にクリックして、パイプラインの実行の状態を確認します。
パイプラインの実行に関連付けられているアクティビティの実行を表示するために、 [アクション] 列の [View Activity Runs](アクティビティの実行の表示) を選択します。

上部の [All Pipeline Runs](すべてのパイプラインの実行) リンクを選択すると、パイプラインの実行ビューに戻ることができます。

出力を検証する

adftutorial コンテナーの spark/otuputfiles/wordcount フォルダーに出力ファイルが作成されていることを確認します。

出力ファイルの場所

このファイルには、入力テキストファイルに含まれている単語と、各単語がファイル内に出現する回数が含まれています。次に例を示します。

(u'This', 1)
(u'a', 1)
(u'is', 1)
(u'test', 1)
(u'file', 1)

このサンプルのパイプラインでは、Spark アクティビティとオンデマンドの HDInsight のリンクされたサービスを使用して、データを変換します。以下の方法を学習しました。

データファクトリを作成します。
Spark アクティビティを使用するパイプラインを作成します。
パイプラインの実行をトリガーする。
パイプラインの実行を監視します。

仮想ネットワークにある Azure HDInsight クラスター上で Hive スクリプトを実行してデータを変換する方法については、次のチュートリアルに進んでください。

チュートリアル: Azure Virtual Network で Hive を使用してデータを変換する。

次の方法で共有

Azure Data Factory で Spark アクティビティを使用してクラウドのデータを変換する