チュートリアル:Azure Data Lake Storage Gen2、Azure Databricks、および Spark

[アーティクル]
10/26/2023

このチュートリアルでは、Azure Data Lake Storage Gen2 対応の Azure ストレージアカウント内の格納データに Azure Databricks クラスターを接続する方法を説明します。この接続を使用することで、必要なデータに関するクエリや分析をクラスターからネイティブに実行することができます。

このチュートリアルでは、次のことについて説明します。

非構造化データをストレージアカウントに取り込む
Blob Storage 内のデータに対して分析を実行する

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

前提条件

階層型名前空間 (Azure Data Lake Storage Gen2) を持つストレージアカウントを作成します

「Azure Data Lake Storage Gen2 で使用するストレージアカウントを作成する」をご覧ください。
ユーザーアカウントにストレージ BLOB データ共同作成者ロールが割り当てられていることを確認します。
AzCopy v10 をインストールします。 AzCopy v10 を使用したデータ転送に関するページを参照してください。
サービスプリンシパルを作成し、クライアントシークレットを作成し、サービスプリンシパルにストレージアカウントへのアクセス権を付与します。

「チュートリアル: Azure Data Lake Storage Gen2 に接続する」 (手順 1 から 3) を参照してください。これらの手順を完了したら、テナント ID、アプリ ID、クライアントシークレットの値をテキストファイルに貼り付けてください。このチュートリアルの後の方で使用します。

Azure Databricks ワークスペース、クラスター、ノートブックを作成する

Azure Databricks ワークスペースを作成する。「Azure Databricks ワークスペースを作成する」をご覧ください。
クラスターを作成する。クラスターの作成に関する記事を参照してください。
Notebook を作成します。「ノートブックを作成する」を参照してください。ノートブックの既定の言語として Python を選択します。

ノートブックを開いたままにします。これは、次のセクションで使用します。

フライトデータのダウンロード

このチュートリアルでは、運輸統計局からの 2016 年 1 月の定刻実績フライトデータを使用して ETL 操作を実行する方法を示します。チュートリアルを完了するには、このデータをダウンロードする必要があります。

On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip ファイルをダウンロードします。このファイルには、フライトデータが含まれています。
ZIP ファイルの内容を解凍し、ファイル名とファイルパスをメモします。この情報は後の手順で必要になります。

定刻レポート実績データでキャプチャされた情報について知りたい方は、運輸統計局の Web サイトでフィールドの説明を参照してください。

データを取り込む

このセクションでは、.csv フライトデータを Azure Data Lake Storage Gen2 アカウントにアップロードし、そのストレージアカウントを Databricks クラスターにマウントします。最後に、Databricks を使用して、.csv フライトデータを読み取り、Apache parquet 形式でストレージに書き戻します。

ストレージアカウントにフライトデータをアップロードする

AzCopy を使用して .csv ファイルを Data Lake Storage Gen2 アカウントにコピーします。 azcopy make コマンドを使用して、ストレージアカウントでコンテナーを作成します。次に、azcopy copy コマンドを使用して、ダウンロードしたばかりの csv データをそのコンテナーのディレクトリにコピーします。

次の手順では、作成するコンテナーの名前と、フライトデータをアップロードする先のコンテナー内のディレクトリと BLOB を入力する必要があります。各手順で推奨される名前を使用するか、コンテナー、ディレクトリ、BLOB の名前付け規則に従って独自の名前を指定できます。

コマンドプロンプトウィンドウを開き、次のコマンドを入力して Azure Active Directory にサインインし、ストレージアカウントにアクセスします。
```
azcopy login
```
コマンドプロンプトウィンドウに表示される指示に従って、ユーザーアカウントを認証します。
フライトデータを保存するコンテナーをストレージアカウントに作成するには、次のコマンドを入力します。
```
azcopy make  "https://<storage-account-name>.dfs.core.windows.net/<container-name>" 
```
- <storage-account-name> プレースホルダーの値は、実際のストレージアカウントの名前に置き換えます。
- <container-name> プレースホルダーを、csv データの保存先コンテナーとして作成するコンテナーの名前、たとえば flight-data-container に置き換えます。
csv データをストレージアカウントにアップロード (コピー) するには、次のコマンドを入力します。
```
azcopy copy "<csv-folder-path>" https://<storage-account-name>.dfs.core.windows.net/<container-name>/<directory-name>/On_Time.csv
```
- プレースホルダー <csv-folder-path> の値は、 .csv ファイルへのパスに置き換えます。
- <storage-account-name> プレースホルダーの値は、実際のストレージアカウントの名前に置き換えます。
- <container-name> プレースホルダーは、お使いのストレージアカウントのコンテナーの名前に置き換えます。
- <directory-name> プレースホルダーは、コンテナー内のデータの保存先ディレクトリの名前、たとえば、jan2016 に置き換えます。

ストレージアカウントを Databricks クラスターにマウントする

このセクションでは、Azure Data Lake Storage Gen2 クラウドオブジェクトストレージを Databricks ファイルシステム (DBFS) にマウントします。ストレージアカウントでの認証には、前に作成した Azure AD サービスプリンシパルを使用します。詳細については、「Azure Databricks へのクラウドオブジェクトストレージのマウント」を参照してください。

クラスターにノートブックをアタッチします。
1. 前に作成したノートブックで、ノートブックツールバーの右上隅にある [接続] ボタン選択します。このボタンをクリックすると、コンピューティングセレクターが開きます (既にノートブックがクラスターに接続されている場合は、[接続] ではなく、そのクラスターの名前のテキストがボタンに表示されます)。
2. クラスターのドロップダウンメニューで、前に作成したクラスターを選択します。
3. クラスターセレクターのテキストが、"開始" に変わることに注意してください。クラスターの開始が完了し、クラスターの名前がボタンに表示されるまで待ってから、先に進みます。

次のコードブロックをコピーして最初のセルに貼り付けます。ただし、このコードはまだ実行しないでください。

configs = {"fs.azure.account.auth.type": "OAuth",
       "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
       "fs.azure.account.oauth2.client.id": "<appId>",
       "fs.azure.account.oauth2.client.secret": "<clientSecret>",
       "fs.azure.account.oauth2.client.endpoint": "https://login.microsoftonline.com/<tenantId>/oauth2/token",
       "fs.azure.createRemoteFileSystemDuringInitialization": "true"}

dbutils.fs.mount(
source = "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<directory-name>",
mount_point = "/mnt/flightdata",
extra_configs = configs)

このコードブロックでは次の操作を行います。
- configs で、<appId>、<clientSecret>、<tenantId> の各プレースホルダーの値を、前提条件でサービスプリンシパルを作成したときにコピーしたアプリケーション ID、クライアントシークレット、テナント ID に置き換えます。
- source URI で、<storage-account-name>、<container-name>、<directory-name> の各プレースホルダーの値を、Azure Data Lake Storage Gen2 ストレージアカウントの名前、およびフライトデータをストレージアカウントにアップロードしたときに指定したコンテナーとディレクトリの名前に置き換えます。
  
  Note
  
  URI のスキーム識別子 (abfss) は、トランスポート層セキュリティ (TLS) で Azure Blob File System ドライバーを使用するように Databricks に指示します。 URI の詳細については、「Azure Data Lake Storage Gen2 の URI を使用する」を参照してください。
クラスターの開始が完了していることを確認してから、次に進みます。
Shift + Enter キーを押して、このブロック内のコードを実行します。

ストレージアカウントでフライトデータをアップロードしたコンテナーとディレクトリに、マウントポイント (/mnt/flightdata) を介してノートブックでアクセスできるようになりました。

Databricks Notebook を使用して CSV を Parquet に変換する

csv フライトデータに DBFS マウントポイントを介してアクセスできるようになりました。これで、Apache Spark DataFrame を使用して、これをワークスペースに読み込み、Apache parquet 形式で Azure Data Lake Storage Gen2 オブジェクトストレージに書き戻すことができます。

Spark DataFrame は、型が異なる可能性のある列を持つ 2 次元のラベル付きデータ構造です。 DataFrame を使用すると、サポートされているさまざまな形式のデータの読み取りと書き込みを容易に行うことができます。 DataFrame を使用すると、クラウドオブジェクトストレージ内の基になるデータに影響を与えることなく、クラウドオブジェクトストレージからデータを読み込み、コンピューティングクラスター内で、それに対して分析と変換を実行できます。詳細については、「チュートリアル: Azure Databricks で PySpark DataFrames を操作する」を参照してください。
Apache parquet は、クエリを高速化する最適化を備えた列形式のファイル形式です。これは CSV や JSON よりも効率的なファイル形式です。詳細については、「Parquet ファイル」を参照してください。

ノートブックで新しいセルを追加し、次のコードを貼り付けます。

# Use the previously established DBFS mount point to read the data.
# Create a DataFrame to read the csv data.
# The header option specifies that the first row of data should be used as the DataFrame column names
# The inferschema option specifies that the column data types should be inferred from the data in the file
flight_df = spark.read.format('csv').options(
    header='true', inferschema='true').load("/mnt/flightdata/*.csv")

# Read the airline csv file and write the output to parquet format for easy query.
flight_df.write.mode("append").parquet("/mnt/flightdata/parquet/flights")
print("Done")

Shift + Enter キーを押して、このブロック内のコードを実行します。

次のセクションに進む前に、すべての parquet データが書き込まれていること、および出力に "完了" と表示されていることを確認します。

データの探索

このセクションでは、前のセクションで作成した DBFS マウントポイントを使用して、Databricks ファイルシステムユーティリティを使って Azure Data Lake Storage Gen2 オブジェクトストレージを探索します。

新しいセルに次のコードを貼り付けて、マウントポイントにあるファイルの一覧を取得します。最初のコマンドによって、ファイルとディレクトリの一覧が出力されます。 2 番目のコマンドによって、出力が読みやすいように表形式で表示されます。

dbutils.fs.ls("/mnt/flightdata")
display(dbutils.fs.ls("/mnt/flightdata"))

Shift + Enter キーを押して、このブロック内のコードを実行します。

parquet ディレクトリが一覧に表示されていることに注意してください。 .csv フライトデータは、前のセクションで parquet/flights ディレクトリに parquet 形式で保存しました。 parquet/flights ディレクトリ内のファイルを一覧表示するには、次のコードを新しいセルに貼り付けて実行します。

display(dbutils.fs.ls("/mnt/flightdata/parquet/flights"))

新しいファイルを作成して表示するには、次のコードを新しいセルに貼り付けて実行します。

dbutils.fs.put("/mnt/flightdata/mydirectory/mysubdirectory/1.txt", "Hello, World!", True)
display(dbutils.fs.ls("/mnt/flightdata/mydirectory/mysubdirectory"))

このチュートリアルでは 1.txt ファイルは不要なので、次のコードをセルに貼り付けて実行し、mydirectory を再帰的に削除できます。 True パラメーターは再帰的な削除を示します。

dbutils.fs.rm("/mnt/flightdata/mydirectory", True)

必要に応じて、ヘルプコマンドを使用して、他のコマンドの詳細を確認できます。

dbutils.fs.help("rm")

上記のコードサンプルでは、Azure Data Lake Storage Gen2 対応のストレージアカウントに格納されたデータを使って HDFS の階層的な性質を調査しました。

データにクエリを実行する

これで、ストレージアカウントにアップロードしたデータの照会を開始できます。次のコードブロックをそれぞれ新しいセルに入力し、Shift + Enter キーを押して Python スクリプトを実行します。

DataFrames には、一般的なデータ分析の問題を効率的に解決できるようにする豊富な機能セット (列の選択、フィルター、結合、集計) が用意されています。

前に保存した parquet フライトデータから DataFrame を読み込み、サポートされている機能の一部を調べるには、このスクリプトを新しいセルに入力して実行します。

# Read the existing parquet file for the flights database that was created earlier
flight_df = spark.read.parquet("/mnt/flightdata/parquet/flights")

# Print the schema of the dataframe
flight_df.printSchema()

# Print the flight database size
print("Number of flights in the database: ", flight_df.count())

# Show the first 25 rows (20 is the default)
# To show the first n rows, run: df.show(n)
# The second parameter indicates that column lengths shouldn't be truncated (default is 20 characters)
flight_df.show(25, False)

# You can also use the DataFrame to run simple queries. Results are returned in a DataFrame.
# Show the first 25 rows of the results of a query that returns selected colums for all flights originating from airports in Texas
flight_df.select("FlightDate", "Reporting_Airline", "Flight_Number_Reporting_Airline", "OriginCityName", "DepTime", "DestCityName", "ArrTime", "ArrDelay").filter("OriginState = 'TX'").show(258, False)

# Use display to run visualizations
# Preferably run this in a separate cmd cell
display(flight_df)

このスクリプトを新しいセルに入力して、データに対して基本的な分析クエリをいくつか実行します。スクリプト全体を実行するか (Shift + Enter)、各クエリを強調表示して Ctrl + Shift + Enter で個別に実行するか、各クエリを個別のセルに入力し、そこで実行できます。

# create a temporary sql view for querying flight information
flight_data = spark.read.parquet('/mnt/flightdata/parquet/flights')
flight_data.createOrReplaceTempView('FlightTable')

# Print the total number of flights in Jan 2016 (the number of rows in the flight data).
print("Number of flights in Jan 2016: ", flight_data.count())

# Using spark sql, query the parquet file to return the total flights of each airline
num_flights_by_airline=spark.sql("SELECT Reporting_Airline, count(*) AS NumFlights FROM FlightTable GROUP BY Reporting_Airline ORDER BY NumFlights DESC")
num_flights_by_airline.show()

# List out all the airports in Texas
airports_in_texas = spark.sql(
    "SELECT DISTINCT(OriginCityName) FROM FlightTable WHERE OriginStateName = 'Texas'")
print('Airports in Texas: ', airports_in_texas.count())
airports_in_texas.show(100, False)

# Find all airlines that fly from Texas
airlines_flying_from_texas = spark.sql(
    "SELECT DISTINCT(Reporting_Airline) FROM FlightTable WHERE OriginStateName='Texas'")
print('Airlines that fly to/from Texas: ', airlines_flying_from_texas.count())
airlines_flying_from_texas.show(100, False)

# List airlines by average arrival delay (negative values indicate early flights)
avg_arrival_delay=spark.sql(
    "SELECT Reporting_Airline, count(*) AS NumFlights, avg(DepDelay) AS AverageDepDelay, avg(ArrDelay) AS AverageArrDelay FROM FlightTable GROUP BY Reporting_Airline ORDER BY AverageArrDelay DESC")
print("Airlines by average arrival delay")
avg_arrival_delay.show()

# List airlines by the highest percentage of delayed flights. A delayed flight is one with a  departure or arrival delay that is greater than 15 minutes
spark.sql("DROP VIEW IF EXISTS totalFlights")
spark.sql("DROP VIEW IF EXISTS delayedFlights")
spark.sql(
    "CREATE TEMPORARY VIEW totalFlights AS SELECT Reporting_Airline, count(*) AS NumFlights FROM FlightTable GROUP BY Reporting_Airline")
spark.sql(
    "CREATE TEMPORARY VIEW delayedFlights AS SELECT Reporting_Airline, count(*) AS NumDelayedFlights FROM FlightTable WHERE DepDelay>15 or ArrDelay>15 GROUP BY Reporting_Airline")
percent_delayed_flights=spark.sql(
    "SELECT totalFlights.Reporting_Airline, totalFlights.NumFlights, delayedFlights.NumDelayedFlights, delayedFlights.NumDelayedFlights/totalFlights.NumFlights*100 AS PercentFlightsDelayed FROM totalFlights INNER JOIN delayedFlights ON totalFlights.Reporting_Airline = delayedFlights.Reporting_Airline ORDER BY PercentFlightsDelayed DESC")
print("Airlines by percentage of flights delayed")
percent_delayed_flights.show()

まとめ

このチュートリアルでは、次の作業を行いました。

Azure Data Lake Storage Gen2 ストレージアカウントや Azure AD サービスプリンシパルなどの Azure リソースを作成し、ストレージアカウントにアクセスするためのアクセス許可を割り当てました。
Azure Databricks ワークスペース、ノートブック、コンピューティングクラスターを作成しました。
AzCopy を使用して、非構造化 .csv フライトデータを Azure Data Lake Storage Gen2 ストレージアカウントにアップロードしました。
Databricks ファイルシステムユーティリティ関数を使用して、Azure Data Lake Storage Gen2 ストレージアカウントをマウントし、その階層型ファイルシステムを探索しました。
Apache Spark DataFrames を使用して、.csv フライトデータを Apache parquet 形式に変換し、それを Azure Data Lake Storage Gen2 ストレージアカウントに保存しました。
DataFrames を使用してフライトデータを探索し、簡単なクエリを実行しました。
Apache Spark SQL を使用してフライトデータに対してクエリを実行し、2016 年 1 月の各航空会社のフライトの合計数、テキサス州の空港、テキサスから運航している航空会社、全国の各航空会社の平均到着遅延時間 (分単位)、出発または到着が遅延した各航空会社のフライトの割合を照会しました。

リソースをクリーンアップする

ノートブックを保持し、後で使用する場合は、課金されないようにクラスターをシャットダウン (終了) することをお勧めします。クラスターを終了するには、それをノートブックツールバーの右上にあるコンピューティングセレクターで選択し、メニューから [終了] を選択して、選択内容を確認します (既定では、クラスターは非アクティブな状態が 120 分続いた後、自動的に終了します)。

ノートブックやクラスターなどの個々のワークスペースリソースを削除する必要がある場合は、ワークスペースの左側のサイドバーから削除できます。詳細な手順については、「クラスターを削除する」または「ノートブックを削除する」を参照してください。

リソースグループおよび関連するすべてのリソースは、不要になったら削除します。 Azure portal でこれを行うには、ストレージアカウントとワークスペースのリソースグループを選択し、[削除] を選択します。

次のステップ

Azure HDInsight の Apache Hive を使用してデータの抽出、変換、読み込みを行う

次の方法で共有

チュートリアル:Azure Data Lake Storage Gen2、Azure Databricks、および Spark

前提条件

Azure Databricks ワークスペース、クラスター、ノートブックを作成する

フライトデータのダウンロード

データを取り込む

ストレージアカウントにフライトデータをアップロードする

ストレージアカウントを Databricks クラスターにマウントする

Databricks Notebook を使用して CSV を Parquet に変換する

データの探索

データにクエリを実行する

まとめ

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

次の方法で共有

チュートリアル:Azure Data Lake Storage Gen2、Azure Databricks、および Spark

前提条件

Azure Databricks ワークスペース、クラスター、ノートブックを作成する

フライト データのダウンロード

データを取り込む

ストレージ アカウントにフライト データをアップロードする

ストレージ アカウントを Databricks クラスターにマウントする

Databricks Notebook を使用して CSV を Parquet に変換する

データの探索

データにクエリを実行する

まとめ

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

フライトデータのダウンロード

ストレージアカウントにフライトデータをアップロードする

ストレージアカウントを Databricks クラスターにマウントする