チュートリアル:Databricks Delta テーブルを更新する Data Lake キャプチャパターンを実装する

このチュートリアルでは、階層型名前空間を持つストレージアカウント内でイベントを処理する方法について説明します。

販売注文を説明するカンマ区切り値(CSV)ファイルをアップロードすることで、DataBricks Deltaテーブルを埋める小さなソリューションを構築します。このソリューションは、Event Gridサブスクリプション、Azure関数、Azure Databricksのジョブを接続することで構築します。

このチュートリアルでは、次の操作を行います。

Azure 関数を呼び出す Event Grid サブスクリプションを追加します。
イベントから通知を受信し、Azure Databricks 内でジョブを実行する Azure 関数を作成します。
ストレージアカウントに配置されている Databricks Delta テーブルに顧客注文を挿入する Databricks ジョブを作成します。

このソリューションは逆順に構築します。まずはAzure Databricksワークスペースから始めます。

前提条件

階層型名前空間 (Azure Data Lake Storage) を持つストレージアカウントを作成します。このチュートリアルでは、contosoorders という名前のストレージアカウントを使用します。

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

「チュートリアル: Azure Data Lake Storage に接続する」 (手順 1 から 3) を参照してください。これらの手順を完了したら、テナントID、アプリID、クライアントシークレットの値をテキストファイルに貼り付けてください。その価値観はすぐに必要です。
Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。

販売注文を作成する

まず、販売注文を説明するCSVファイルを作成し、そのファイルをストレージアカウントにアップロードします。後で、このファイルのデータを使い、Databricks Deltaテーブルの最初の行を埋めます。

Azure ポータルで新しいストレージアカウントに移動します。
Storage browser>Blob containers>Add containerを選択し、新しいコンテナを「data」と名付けて作成します。
data コンテナーで、input という名前のディレクトリを作成します。

次のテキストをテキストエディターに貼り付けます。

InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country
536365,85123A,WHITE HANGING HEART T-LIGHT HOLDER,6,12/1/2010 8:26,2.55,17850,United Kingdom

このファイルをローカルのパソコンに保存し、名前を data.csvと名付けてください。
ストレージブラウザでこのファイルを入力フォルダにアップロードします。

Azure Databricks でジョブを作成する

このセクションでは、以下の作業を行います:

Azure Databricks ワークスペースを作成する。
Notebook を作成します。
Databricks Delta テーブルを作成し、データを設定します。
Databricks Delta テーブルに行を挿入するコードを追加します。
ジョブを作成する。

Azure Databricks ワークスペースを作成する

このセクションでは、Azureポータルを使ってAzure Databricksワークスペースを作成します。

Azure Databricks ワークスペースを作成する。作業スペースの名前は contoso-orders。「Azure Databricks ワークスペースを作成する」をご覧ください。
クラスターを作成する。クラスターに customer-order-cluster という名前を付けます。クラスターの作成に関する記事を参照してください。
Notebook を作成します。ノートブックに configure-customer-table という名前を付け、そのノートブックの既定の言語として Python を選択します。「ノートブックを作成する」を参照してください。

Databricks Delta テーブルを作成してデータを設定する

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

このコードブロックの appId、 password、 tenant のプレースホルダーの値を、このチュートリアルの前提条件をクリアした際に集めた値に置き換えてください。
```
dbutils.widgets.text('source_file', "", "Source File")

spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id", "<appId>")
spark.conf.set("fs.azure.account.oauth2.client.secret", "<password>")
spark.conf.set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<tenant>/oauth2/token")

adlsPath = 'abfss://data@contosoorders.dfs.core.windows.net/'
inputPath = adlsPath + dbutils.widgets.get('source_file')
customerTablePath = adlsPath + 'delta-tables/customers'
```
このコードによって、source_file という名前のウィジェットが作成されます。後で、このコードを呼び出し、そのウィジェットにファイルパスを渡す Azure 関数を作成します。このコードはストレージアカウントでサービスプリンシパルを認証し、他のセルで使う変数も作成します。

注

運用設定では、認証キーを Azure Databricks に格納することを検討してください。そして、認証キーの代わりにルックアップキーをコードブロックに追加します。

例えば、このコード行 spark.conf.set("fs.azure.account.oauth2.client.secret", "<password>")の代わりに、次のコード行を使うと spark.conf.set("fs.azure.account.oauth2.client.secret", dbutils.secrets.get(scope = "<scope-name>", key = "<key-name-for-service-credential>"))。

このチュートリアルを終えた後は、Azure DatabricksのウェブサイトにあるAzure Data Lake Storageの記事でこのアプローチの例をご覧ください。
SHIFT+ENTERを押して、このブロック内のコードを実行してください。

次のコードブロックを別のセルにコピー&ペーストし、 SHIFT+ENTER を押してそのブロック内のコードを実行します。

from pyspark.sql.types import StructType, StructField, DoubleType, IntegerType, StringType

inputSchema = StructType([
StructField("InvoiceNo", IntegerType(), True),
StructField("StockCode", StringType(), True),
StructField("Description", StringType(), True),
StructField("Quantity", IntegerType(), True),
StructField("InvoiceDate", StringType(), True),
StructField("UnitPrice", DoubleType(), True),
StructField("CustomerID", IntegerType(), True),
StructField("Country", StringType(), True)
])

rawDataDF = (spark.read
 .option("header", "true")
 .schema(inputSchema)
 .csv(adlsPath + 'input')
)

(rawDataDF.write
  .mode("overwrite")
  .format("delta")
  .saveAsTable("customer_data", path=customerTablePath))

このコードはストレージアカウント内のDatabricks Deltaテーブルを作成し、先にアップロードしたCSVファイルから初期データを読み込みます。

このコードブロックが正常に実行されたら、ノートブックからこのコードブロックを削除してください。

Databricks Delta テーブルに行を挿入するコードを追加する

次のコードブロックをコピーして、別のセルに貼り付けます。ただし、このセルは実行しないでください。
```
upsertDataDF = (spark
  .read
  .option("header", "true")
  .csv(inputPath)
)
upsertDataDF.createOrReplaceTempView("customer_data_to_upsert")
```
このコードは、CSVファイルからのデータを使って一時的なテーブルビューにデータを挿入します。そのCSVファイルへのパスは、前のステップで作成した入力ウィジェットから来ています。

次のコードブロックをコピーして別のセルに貼り付けます。このコードにより、一時テーブルビューの内容を Databricks Delta テーブルとマージします。

%sql
MERGE INTO customer_data cd
USING customer_data_to_upsert cu
ON cd.CustomerID = cu.CustomerID
WHEN MATCHED THEN
  UPDATE SET
    cd.StockCode = cu.StockCode,
    cd.Description = cu.Description,
    cd.InvoiceNo = cu.InvoiceNo,
    cd.Quantity = cu.Quantity,
    cd.InvoiceDate = cu.InvoiceDate,
    cd.UnitPrice = cu.UnitPrice,
    cd.Country = cu.Country
WHEN NOT MATCHED
  THEN INSERT (InvoiceNo, StockCode, Description, Quantity, InvoiceDate, UnitPrice, CustomerID, Country)
  VALUES (
    cu.InvoiceNo,
    cu.StockCode,
    cu.Description,
    cu.Quantity,
    cu.InvoiceDate,
    cu.UnitPrice,
    cu.CustomerID,
    cu.Country)

ジョブの作成

以前作成したノートブックを動かすジョブを作成しましょう。その後、イベントが発生した際にこのジョブを実行するためのAzure関数を作成します。

「新しく>仕事」を選択してください。
ジョブに名前をつけ、作成したノートブックを選び、クラスタを選択します。次に、[作成] を選択してジョブを作成します。

新しい仕事は、選択したノートブックとクラスタと一緒に ジョブリスト に表示されます。

Azure Function の作成

ジョブを実行するAzure関数を作成します。

Azure Databricks workspacesで、上部バーでAzure Databricksのユーザー名を選択してください。ドロップダウンリストから 「ユーザー設定」を選択してください。
[アクセストークン] タブで、[新しいトークンの生成] を選択します。
表示されたトークンをコピーし、「完了」を選択します。
Databricks ワークスペースの上隅で、人のアイコンを選択し、 [ユーザー設定] を選択します。
[新しいトークンの生成] を選択し、[生成] を選択します。

トークンは必ず安全な場所にコピーしてください。 Azure関数はこのトークンを必要とし、Databricksで認証してジョブを実行する必要があります。
Azure portal のメニューまたは [ホーム] ページから [リソースの作成] を選択します。
[新規] ページで、 [計算]>、 [関数アプリ] の順に選択します。

[関数アプリを作成する] ページの [基本] タブで、リソースグループを選択し、次の設定を変更または確認します。

設定	値
関数アプリ名	contosoorder
ランタイムスタック	.NET
公開	Code
オペレーティングシステム	Windows
プランの種類	従量課金 (サーバーレス)

[確認と作成] を選択し、次に [作成] を選択します。

デプロイが完了したら、[リソースに移動] を選択して関数アプリの概要ページを開きます。
[設定] グループで、[構成] を選択します。

[アプリケーションの設定] ページで、 [新しいアプリケーション設定] を選択して各設定を追加します。

Function Appの設定に新しいアプリケーション設定を追加するスクリーンショットです。

以下の設定を追加します。

設定名	値
DBX_INSTANCE	Databricks ワークスペースのリージョンです。例: `westus2.azuredatabricks.net`
DBX_PAT	前に生成した個人用アクセストークン。
DBX_JOB_ID	実行中のジョブの識別子。

[保存] を選択して、これらの設定をコミットします。
[関数] グループで [関数] を選択し、[作成] を選択します。
Azure Event Grid トリガー を選択します。

Microsoft.Azure.WebJobs.Extensions.EventGrid 拡張機能のインストールを求めるメッセージが表示されたら、インストールします。インストールが必要な場合は、再度Azure Event Grid Triggerを選択して関数を作成します。

[新しい関数] ウィンドウが表示されます。
新しい関数で関数名をUpsertOrderを入力し、その後「作成」を選択します。

コードファイルの内容を次のコードに置き換え、その後 「保存」を選択します。

  #r "Azure.Messaging.EventGrid"
  #r "System.Memory.Data"
  #r "Newtonsoft.Json"
  #r "System.Text.Json"
  using Azure.Messaging.EventGrid;
  using Azure.Messaging.EventGrid.SystemEvents;
  using Newtonsoft.Json;
  using Newtonsoft.Json.Linq;

  private static HttpClient httpClient = new HttpClient();

  public static async Task Run(EventGridEvent eventGridEvent, ILogger log)
  {
     log.LogInformation("Event Subject: " + eventGridEvent.Subject);
     log.LogInformation("Event Topic: " + eventGridEvent.Topic);
     log.LogInformation("Event Type: " + eventGridEvent.EventType);
     log.LogInformation(eventGridEvent.Data.ToString());

     if (eventGridEvent.EventType == "Microsoft.Storage.BlobCreated" || eventGridEvent.EventType == "Microsoft.Storage.FileRenamed") {
        StorageBlobCreatedEventData fileData = eventGridEvent.Data.ToObjectFromJson<StorageBlobCreatedEventData>();
        if (fileData.Api == "FlushWithClose") {
              log.LogInformation("Triggering Databricks Job for file: " + fileData.Url);
              var fileUrl = new Uri(fileData.Url);
              var httpRequestMessage = new HttpRequestMessage {
                 Method = HttpMethod.Post,
                 RequestUri = new Uri(String.Format("https://{0}/api/2.0/jobs/run-now", System.Environment.GetEnvironmentVariable("DBX_INSTANCE", EnvironmentVariableTarget.Process))),
                 Headers = { 
                    { System.Net.HttpRequestHeader.Authorization.ToString(), "Bearer " + System.Environment.GetEnvironmentVariable("DBX_PAT", EnvironmentVariableTarget.Process)},
                    { System.Net.HttpRequestHeader.ContentType.ToString(), "application/json" }
                 },
                 Content = new StringContent(JsonConvert.SerializeObject(new {
                    job_id = System.Environment.GetEnvironmentVariable("DBX_JOB_ID", EnvironmentVariableTarget.Process),
                    notebook_params = new {
                          source_file = String.Join("", fileUrl.Segments.Skip(2))
                    }
                 }))
              };
              var response = await httpClient.SendAsync(httpRequestMessage);
              response.EnsureSuccessStatusCode();
        }
     }
  }

このコードは、発生したストレージイベントの情報を解析し、そのイベントをトリガーしたファイルのURLを含むリクエストメッセージを作成します。この関数は、前に作成した source_file ウィジェットに、メッセージの一部として値を渡します。関数コードはメッセージをDatabricksジョブに送り、以前取得したトークンを認証として使います。

Event Grid のサブスクリプションを作成する

このセクションでは、ファイルがストレージアカウントにアップロードされた際に Azure 関数を呼び出すイベントグリッドのサブスクリプションを作成します。

[統合] を選択します統合ページでイベントグリッドトリガーを選択してください。
[トリガーの編集] ウィンドウで、イベントに eventGridEvent という名前を付け、[イベントサブスクリプションの作成] を選択します。

注

名前eventGridEvent Azure関数が受け取るパラメータ名と一致します。

[イベントサブスクリプションの作成] ページの [基本] タブで、次の設定を変更または確認します。

設定	値
Name	contoso-order-event-subscription
トピックの種類	ストレージアカウント
ソースリソース	contosoorders
[システムトピック名]	`<create any name>`
イベントの種類のフィルター	BLOB の作成と削除

を選択してを作成します。

Event Grid サブスクリプションをテストする

customer-order.csv という名前のファイルを作成し、そのファイルに次の情報を貼り付けて、ローカルコンピューターに保存します。
```
InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country
536371,99999,EverGlow Single,228,1/1/2018 9:01,33.85,20993,Sierra Leone
```
ストレージブラウザで、このファイルをストレージアカウントの入力フォルダにアップロードしてください。

ファイルをアップロードすると、Microsoft.Storage.BlobCreated イベントが発生します。 Event Grid は、そのイベントのすべてのサブスクライバーに通知を発信します。この場合、Azure関数だけが加入者となります。 Azure 関数は、イベントパラメーターを解析して、発生したイベントを特定します。その後、ファイルのURLをDatabricksジョブに渡します。 Databricksジョブはファイルを読み取り、ストレージアカウントにあるDatabricks Deltaテーブルに行を追加します。
ジョブが成功したかどうかを確認するには、ジョブの実行履歴を確認してください。完成状態が表示されます。作業の実行表示方法については、「作業の実行表示」をご覧ください。
新しいワークブックのセルでこのクエリを実行し、更新されたデルタテーブルを確認します。
```
%sql select * from customer_data
```
返されるテーブルには、最新のレコードが表示されます。
このレコードを更新するには、customer-order-update.csv という名前のファイルを作成し、そのファイルに次の情報を貼り付けて、ローカルコンピューターに保存します。
```
InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country
536371,99999,EverGlow Single,22,1/1/2018 9:01,33.85,20993,Sierra Leone
```
このCSVファイルは前のものとほぼ同じですが、注文の数量が 228 から 22に変更されています。
ストレージブラウザで、このファイルをストレージアカウントの入力フォルダにアップロードしてください。
select クエリを再実行して、更新された Delta テーブルを表示します。
```
%sql select * from customer_data
```
返されるテーブルには、更新後のレコードが表示されます。

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

リソースが不要になったら、リソースグループとすべての関連リソースを削除します。リソースグループを削除するには、ストレージアカウントのリソースグループを選択し、「削除」を選択します。

次のステップ

Blob Storage のイベント処理

フィードバック

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

Last updated on 2026-07-25