Azure Data Factory でカスタム イベント トリガーを作成してパイプラインを実行する
適用対象: 
Azure Data Factory Azure Synapse Analytics
ヒント
企業向けのオールインワン分析ソリューション、Microsoft Fabric の Data Factory をお試しください。 Microsoft Fabric は、データ移動からデータ サイエンス、リアルタイム分析、ビジネス インテリジェンス、レポートまで、あらゆるものをカバーしています。 無料で新しい試用版を開始する方法について説明します。
Azure Data Factory のイベント トリガーを使用すると、データ ソースで発生する特定のイベントに基づいて、パイプラインの実行を自動化できます。 これはイベント ドリブン アーキテクチャの重要な機能であり、リアルタイムのデータ統合と処理を可能にするものです。
イベントドリブン アーキテクチャ は、イベントの運用、検出、使用、およびリアクションを伴う、データ統合の共通的なパターンです。 多くの場合、データ統合シナリオでは、特定のイベントが発生したときに Azure Data Factory ユーザーがパイプラインをトリガーする必要があります。 Data Factory と Azure Event Grid がネイティブに統合され、カスタム トピックがカバーされました。 Azure Event Grid トピックにイベントを送信します。 Data Factory では、トピックをサブスクライブしてリッスンし、それに応じてパイプラインをトリガーします。
この記事で説明されている統合は、Azure Event Grid に依存しています。 サブスクリプションが Event Grid リソース プロバイダーに登録されていることを確認してください。 詳細については、「リソース プロバイダーと種類」を参照してください。 Microsoft.EventGrid/eventSubscriptions/ アクションを実行できる必要があります。 このアクションは、EventGrid EventSubscription 共同作成者の組み込みロールの一部です。
重要
Azure Synapse Analytics でこの機能を使用している場合は、お持ちのサブスクリプションが、Data Factory リソース プロバイダーにも登録されていることを確認してください。 これ以外では、"イベント サブスクリプション が作成できませんでした" というメッセージが表示されます。
パイプライン パラメーターとカスタム イベント トリガーを組み合わせると、パイプラインの実行でカスタム data ペイロードを解析して参照できます。 カスタム イベント ペイロードの data フィールドは自由形式の JSON キー値構造なので、イベント駆動型パイプラインの実行を制御できます。
重要
パラメーター化で参照されるキーがカスタム イベント ペイロードに存在しない場合、trigger run は失敗します。 keyName プロパティが存在しないため、式を評価できないことを示すメッセージが表示されます。 この場合、イベントによって no pipeline run がトリガーされます。
イベントとトリガーのユース ケース
トリガーは、次のようなさまざまなイベントによって起動される可能性があります。
Blob Created: 指定したコンテナーに新しいファイルがアップロードされたとき。 Blob Deleted: コンテナーからファイルが削除されたとき。 Blob Modified: 既存のファイルが更新されたとき。
イベントを使用して、パイプラインの実行を動的にコントロールできます。 たとえば、新しいデータ ファイルが Azure Blob Storage の "incoming" フォルダーにアップロードされると、トリガーによってそのデータを処理するパイプラインが自動的に開始され、タイムリーなデータ統合が確実に行われます。
Event Grid でカスタム トピックを設定する
Data Factory でカスタム イベント トリガーを使用するには、"まず" Event Grid でカスタム トピックを設定する必要があります。
Event Grid に移動し、自分でトピックを作成します。 カスタム トピックを作成する方法の詳細については、Event Grid のポータルのチュートリアルと Azure CLI のチュートリアルに関する記事を参照してください。
Note
このワークフローは、ストレージ イベント トリガーとは異なります。 ここでは、トピックは Data Factory によって設定されません。
Data Factory は、イベントが Event Grid イベント スキーマに従うことを想定しています。 イベント ペイロードに、次のフィールドがあることを確認してください。
[
{
"topic": string,
"subject": string,
"id": string,
"eventType": string,
"eventTime": string,
"data":{
object-unique-to-each-publisher
},
"dataVersion": string,
"metadataVersion": string
}
]
Data Factory を使用してカスタム イベント トリガーを作成する
Data Factory に移動してサインインします。
[編集] タブに切り替えます。鉛筆アイコンを探します。
メニューの [トリガー] を選択し、 [新規作成/編集] を選択します。
[トリガーの追加] ページで、[トリガーの選択] を選択した後、[+ 新規] を選択します。
[種類] に [カスタム イベント] を選択します。
[Azure サブスクリプション] ドロップダウン リストからカスタム トピックを選択するか、イベント トピックのスコープを手動で入力します。
Note
Data Factory でカスタム イベント トリガーを作成または変更するには、適切な Azure ロールベースのアクセス制御 (Azure RBAC) を持つ Azure アカウントを使用する必要があります。 他のアクセス許可は必要ありません。 Data Factory のサービス プリンシパルでは、Event Grid に対する特別なアクセス許可は必要ありません。 アクセス制御の詳細については、「ロールベースのアクセス制御」セクションを参照してください。
Subject begins withプロパティとSubject ends withプロパティを使用すると、トリガー イベントをフィルター処理できます。 これらのプロパティは両方とも省略可能です。[+ 新規] を使用して、フィルター処理する [イベントの種類] を追加します。 カスタム イベント トリガーの一覧では、OR リレーションシップが使用されます。
eventTypeプロパティを持つカスタム イベントが一覧上の 1 つと一致すると、パイプラインの実行がトリガーされます。 イベントの種類では大文字と小文字が区別されます。 たとえば、次のスクリーンショットでは、トリガーは、サブジェクトが factories で始まるすべてのcopycompletedまたはcopysucceededイベントと一致します。![Data Factory UI でのイベントの種類とサブジェクトのフィルター処理を説明する [トリガーの編集] ページを示すスクリーンショット。](media/how-to-create-custom-event-trigger/custom-event-2-properties.png)
カスタム イベント トリガーを使用して、カスタム
dataペイロードを解析し、パイプラインに送信できます。 パイプライン パラメーターを作成し、[パラメーター] ページで値を入力します。 形式@triggerBody().event.data._keyName_を使用してデータ ペイロードを解析し、パイプラインのパラメーターに値を渡します。詳細については、次を参照してください。
パラメーターを入力したら、[OK] を選択します。
高度なフィルター処理
カスタム イベント トリガーでは、Event Grid の高度なフィルター処理と同様の高度なフィルター処理機能がサポートされています。 これらの条件付きフィルターを使用すると、イベント ペイロードの値に基づいてパイプラインをトリガーできます。 たとえば、イベント ペイロードに Department という名前のフィールドがあり、Department が Finance である場合にのみ、パイプラインをトリガーする必要があるとします。 date フィールドにリスト [1, 2, 3, 4, 5]、month フィールドの not 条件にリスト [11, 12]、tag フィールドに[Fiscal Year 2021, FiscalYear2021, or FY2021] が含まれるかどうかなど、複雑なロジックも指定できます。
カスタム イベント トリガーでは現在、高度なフィルター処理演算子の一部が Event Grid でサポートされています。 次のフィルター条件がサポートされています。
NumberInNumberNotInNumberLessThanNumberGreaterThanNumberLessThanOrEqualsNumberGreaterThanOrEqualsBoolEqualsStringContainsStringBeginsWithStringEndsWithStringInStringNotIn
[+ 新規] を選択して、新しいフィルター条件を追加します。
カスタム イベント トリガーは、Event Grid と同じ制限にも従います。次に例を示します。
- カスタム イベント トリガーごとに、すべてのフィルターで 5 つの高度なフィルターと 25 のフィルター値。
- 文字列値あたり 512 文字。
in演算子とnot in演算子の 5 つの値。john.doe@contoso.comなど、キーに.(ドット) 文字を含めることはできません。 現時点では、エスケープ文字を含むキーはサポートされていません。- 複数のフィルターで同じキーを使用できます。
Data Factory は、Event Grid API の最新の一般提供 (GA) バージョンに依存します。 新しい API バージョンが GA 段階になると、Data Factory では、より高度なフィルター処理演算子のサポートが拡張されます。
JSON スキーマ
次の表に、カスタム イベント トリガーに関連するスキーマ要素の概要を示します。
| JSON 要素 | 説明 | Type | 使用できる値 | 必須 |
|---|---|---|---|---|
scope |
イベント グリッド トピックの Azure Resource Manager リソース ID。 | String | Azure Resource Manager ID | はい。 |
events |
このトリガーを起動するイベントの種類。 | 文字列の配列 | はい。少なくとも 1 つの値が必要です。 | |
subjectBeginsWith |
subject フィールドは、トリガーを起動するために指定されたパターンで始まる必要があります。 たとえば、factories は、factories で始まるイベント サブジェクトに対してのみトリガーを発動させます。 |
String | いいえ。 | |
subjectEndsWith |
subject フィールドは、トリガーを起動するために指定されたパターンで終わる必要があります。 |
String | いいえ。 | |
advancedFilters |
それぞれフィルター条件を指定する JSON BLOB のリスト。 各 BLOB は key、operatorType、values を指定します。 |
JSON BLOB のリスト | いいえ。 |
ロールベースのアクセス制御
Data Factory では、承認されていないアクセスを禁止するために Azure RBAC が使用されます。 正常に機能するには、Data Factory で、以下を行うためのアクセス権が必要です。
- イベントをリッスンする。
- イベントから更新をサブスクライブする。
- カスタム イベントにリンクされているパイプラインをトリガーする。
カスタム イベント トリガーを正常に作成または更新するには、適切なアクセス権を持つ Azure アカウントを使用して Data Factory にサインインする必要があります。 それ以外の場合、"アクセスが拒否されました" というメッセージで操作が失敗します。
Data Factory では、Event Grid のインスタンスに対する特別なアクセス許可は必要ありません。 この操作のために Data Factory サービス プリンシパルに特別な Azure RBAC ロールのアクセス許可を割り当てる必要も "ありません"。
具体的には、/subscriptions/####/resourceGroups//####/providers/Microsoft.EventGrid/topics/someTopics に対する Microsoft.EventGrid/EventSubscriptions/Write 権限が必要です。
- データ ファクトリ (たとえば開発環境内など) で作成する場合は、サインインしている Azure アカウントに前述のアクセス許可が必要です。
- 継続的インテグレーションと継続的デリバリーを通じて発行する場合、Azure Resource Manager テンプレートをテストまたは運用ファクトリに発行するために使用されるアカウントには、前述のアクセス許可が必要です。
関連するコンテンツ
- トリガー実行に関する詳細情報を取得する。
- パイプライン実行でのトリガー メタデータの参照方法について学ぶ。