Eventstream にレイクハウスの転送先を追加する

[アーティクル]
05/21/2024

この記事では、Microsoft Fabric イベントストリームのイベントストリームに転送先としてレイクハウスを追加する方法について説明します。

Note

プレビュー段階の拡張機能を使用する場合は、上部にある [拡張機能] を選択します。それ以外の場合は、[Standard機能] を選択します。プレビュー段階の拡張機能の詳細については、「Fabric イベントストリームの概要」を参照してください。

重要

レイクハウスの転送先テーブルに、データを書き込むためのスキーマの適用があります。データ品質を確保するために、テーブルへの新しい書き込みはすべて、書き込み時にターゲットテーブルのスキーマとの互換性が必要です。

新しい Delta テーブルに出力を書き込むときは、最初のレコードに基づいてテーブルスキーマが作成されます。出力データのすべてのレコードは、既存のテーブルのスキーマに投影されます。

受信データに既存のテーブルスキーマにない列がある場合、追加の列はテーブルに書き込まれるデータに含まれません。逆に、受信データに、既存のテーブルスキーマにある列が足りていない場合は、その列の値が null としてテーブルに書き込まれます。

前提条件

イベントストリームが配置されている Fabric の Premium ワークスペースに、寄稿者以上のアクセス許可を使用してアクセスします。
レイクハウスが配置されている Premium ワークスペースに、寄稿者以上のアクセス許可を使用してアクセスします。

Note

1 つのイベントストリームでのソースと宛先の合計数は 11 個までです。

Lakehouseを転送先として追加する

既定または派生イベントストリームにレイクハウスの転送先を追加するには、次の手順に従います。

イベントストリームの編集モードで、リボンの [宛先の追加] を選択し、ドロップダウンリストから [レイクハウス] を選択します。
レイクハウスノードをストリームノードまたはオペレーターに接続します。
レイクハウスの構成画面で、次の情報を入力します。
1. [宛先名] を入力します。
2. レイクハウスを含むワークスペースを選択します。
3. 指定したワークスペースから既存のレイクハウスを選択します。
4. 既存の Delta テーブルを選択するか、新しい Delta テーブルを作成してデータを受信します。
5. レイクハウスに送信される入力データの形式を選択します。サポートされているデータ形式は、JSON、Avro、CSV (ヘッダー付き) です。
[Advanced] \(詳細設定) を選択します。
Lakehouseの変換先の場合、2 つのインジェストモードを使用できます。実際のシナリオに基づいて Fabric のイベントストリームによるレイクハウスへの書き込み方法を最適化するために、これらのモードを構成します。
- [最小行] は、レイクハウスでひとつのファイルに取り込む最小行数です。最小値は 1 行で、最大値はファイルあたり最大 200 万行です。最小行数が小さいほど、インジェスト中にレイクハウスによって作成されるファイルが多くなります。
- [最大期間 は、レイクハウスでひとつのファイルを取り込むのにかける最大時間です。最小値は 1 分、最大値は 2 時間です。この期間が長いほど、より多くの行がファイルに取り込まれます。
[保存] を選択します。
新しく追加されたレイクハウスの転送先を実装するには、[発行] を選択します。

これらの手順を完了すると、レイクハウスの転送先をライブビューで視覚化できるようになります。 [詳細] ウィンドウで、ノートブックのテーブルを最適化ショートカットを選択して、ノートブック内で Apache Spark ジョブを起動できます。これにより、ターゲットのレイクハウステーブル内の小さなストリーミングファイルを統合します。

Eventstream に他の転送先を追加する方法については、次の記事を参照してください。

前提条件

開始する前に、次の前提条件を満たしている必要があります。

イベントストリームが配置されている Premium ワークスペースに、寄稿者以上のアクセス許可を使用してアクセスします。
Lakehouse が配置されている Premium ワークスペースに対して、寄稿者以上のアクセス許可を取得します。

Note

1 つのイベントストリームでのソースと宛先の合計数は 11 個までです。

Lakehouseを転送先として追加する

ワークスペースにLakehouseを作成してある場合は、次の手順に従って、そのLakehouseを宛先としてイベントストリームに追加します。

リボン上の [新しい転送先] またはメインエディターのキャンバス内の [+] を選択し、[Lakehouse] を選択します。宛先となるLakehouseを構成する画面が表示されます。
イベントストリームにおける宛先の名前を入力し、Lakehouseに関する情報をすべて入力します。
1. [Lakehouse]: 指定したワークスペースから既存のLakehouse アイテムを選択します。
2. [Delta テーブル]: 既存の Delta テーブルを選択するか、新しい Delta テーブルを作成してデータを受信します。
  
  Note
  
  Lakehouse テーブルにデータを書き込む場合は、 スキーマが適用されます。これはつまり、データ品質を確保するために、テーブルへの新しい書き込みはすべて、書き込み時にターゲットテーブルのスキーマとの互換性が必要であるということです。
  
  出力データのすべてのレコードは、既存のテーブルのスキーマに投影されます。新しい Delta テーブルに出力を書き込むときは、最初のレコードに基づいてテーブルスキーマが作成されます。受信データに、既存のテーブルスキーマと比較して余分な列がある場合は、テーブルへの書き込みではその余分な列は含まれません。逆に、受信データに、既存のテーブルスキーマと比較して列が足りていない場合は、その列の値が null としてテーブルに書き込まれます。
3. [入力データ形式]: Lakehouseに送信されるデータ (入力データ) の形式を選択します。
  
  Note
  
  サポートされている入力イベントデータ形式は、JSON、Avro、CSV (ヘッダー付き) です。
4. [イベント処理]: イベント処理エディターを使用して、レイクハウスに送信する前のデータの処理方法を指定できます。 [イベントプロセッサを開く] を選択して、イベント処理エディターを開きます。イベントプロセッサを使用したリアルタイム処理について詳しくは、「イベントプロセッサエディターを使用してイベントデータを処理する」をご覧ください。エディターを使用し終えたら、[完了] を選択して、宛先となるLakehouseを構成する画面に戻ります。
Lakehouseの変換先の場合、2 つのインジェストモードを使用できます。これらのモードのいずれかを選択して、実際のシナリオに基づいて Fabric のイベントストリーム機能によるLakehouseへの書き込み方法を最適化します。
1. [ファイルあたりの行数] - Lakehouse でひとつのファイルに取り込む最小行数。最小行数が小さいほど、インジェスト中にレイクハウスによって作成されるファイルはより多くなります。最小値は 1 です。ファイルあたりの最大行数は 2M です。
2. [期間] - Lakehouse でひとつのファイルを取り込むのにかける最大時間。この期間が長いほど、より多くの行がファイルに取り込まれます。最小値は 1 分、最大値は 2 時間です。
[追加] を選択し、転送先としてLakehouseを追加します。
Lakehouseの変換先内で使用できるテーブル最適化ショートカット。このソリューションを使用すると、Notebook 内で Spark ジョブを起動し、これらの小さなストリーミングファイルをターゲット Lakehouse テーブル内に統合することが容易になります。
キャンバスに宛先としてLakehouseが表示され、スピン状態インジケーターが表示されます。状態が [アクティブ] に変わるまでに数分かかります。