この記事では、Databricks Lakeflow Connect を使用して ServiceNow インジェスト パイプラインを作成する方法について説明します。
開始する前に
インジェスト パイプラインを作成するには、以下の要件を満たす必要があります。
ワークスペースは、Unity Catalog に対して有効にする必要があります。
ワークスペースに対してサーバーレス コンピューティングを有効にする必要があります。 「サーバーレス コンピューティングを有効にする」をご覧ください。
新しい接続を作成する場合: メタストアに対する
CREATE CONNECTION特権が必要です。コネクタで UI ベースのパイプライン作成がサポートされている場合は、このページの手順を完了することで、接続とパイプラインを同時に作成できます。 ただし、API ベースのパイプライン作成を使用する場合は、このページの手順を完了する前に、カタログ エクスプローラーで接続を作成する必要があります。 「マネージド インジェスト ソースへの接続」を参照してください。
既存の接続を使用する場合: 接続オブジェクトに対する
USE CONNECTION特権またはALL PRIVILEGESが必要です。ターゲット カタログに対する
USE CATALOG特権が必要です。既存のスキーマに対する
USE SCHEMA権限とCREATE TABLE権限、またはターゲット カタログに対するCREATE SCHEMA権限が必要です。
ServiceNow から取り込むには、「 Databricks のインジェスト用に ServiceNow を構成する」を参照してください。
インジェスト パイプラインを作成する
必要なアクセス許可:USE CONNECTION または ALL PRIVILEGES は接続上で必要です。
このステップでは、インジェスト パイプラインを作成する方法について説明します。 取り込まれた各テーブルは、同じ名前のストリーミング テーブルに書き込まれます。
Databricks ユーザーインターフェース
Azure Databricks ワークスペースのサイドバーで、[ Data Ingestion をクリックします。
[ データの追加 ] ページの [Databricks コネクタ] で、[ ServiceNow] をクリックします。
インジェスト ウィザードが開きます。
ウィザードの [ インジェスト パイプライン ] ページで、パイプラインの一意の名前を入力します。
[ 宛先カタログ ] ドロップダウン メニューで、カタログを選択します。 取り込まれたデータとイベント ログがこのカタログに書き込まれます。 後で宛先スキーマを選択します。
ソース データへのアクセスに必要な資格情報を格納する Unity カタログ接続を選択します。
ソースへの既存の接続がない場合は、[接続の 作成 ] をクリックし、「 ServiceNow for Databricks インジェストの構成」で取得した認証の詳細を入力します。 メタストアに対する
CREATE CONNECTION特権が必要です。[パイプラインの作成] をクリックして続行します。
[ ソース ] ページで、取り込むテーブルを選択し、[ 次へ] をクリックします。
[すべてのテーブル] を選択すると、コネクタはソース スキーマ内のすべての既存のテーブルと将来のテーブルを宛先スキーマに書き込みます。 パイプラインあたり最大 250 個のテーブルがあります。
[ 宛先 ] ページで、書き込む Unity カタログ カタログとスキーマを選択します。
既存のスキーマを使用しない場合は、[ スキーマの作成] をクリックします。 親カタログに対する
USE CATALOGおよびCREATE SCHEMA特権が必要です。[パイプラインの保存] をクリックして続行。
(省略可能)[ 設定] ページで、[ スケジュールの作成] をクリックします。 変換先テーブルを更新する頻度を設定します。
(省略可能)パイプライン操作の成功または失敗に関する電子メール通知を設定します。
[ 保存してパイプラインを実行] をクリックします。
Databricks ノートブック
後でノートブックに貼り付けることができるように、個人用アクセス トークンを生成し、トークンをコピーします。 ワークスペース ユーザーの Azure Databricks 個人用アクセス トークンを参照してください。
次のノートブックをワークスペースにインポートします。
ServiceNow インジェスト パイプラインを作成する
ノートブックで次の値を変更します。
セル 1:
-
api_token: 生成した個人用アクセス トークン
セル 3:
name: パイプラインの名前connection_name: カタログ エクスプローラーで作成した Unity カタログ接続の名前 (カタログ > 外部データ > 接続)。 ソースへの既存の接続がない場合は、作成できます。 メタストアに対するCREATE CONNECTION特権が必要です。source_table: ソース テーブルの名前destination_catalog: 取り込まれたデータを格納する宛先カタログの名前。destination_schema: 取り込まれたデータを含む宛先スキーマの名前。scd_type: 使用する SCD メソッド:SCD_TYPE_1またはSCD_TYPE_2。include_columns: 必要に応じて、インジェストに含める列の一覧を指定します。 このオプションを使用して列を明示的に含める場合、パイプラインは将来ソースに追加される列を自動的に除外します。 今後の列を取り込むには、それらを一覧に追加する必要があります。exclude_columns: 必要に応じて、インジェストから除外する列の一覧を指定します。 このオプションを使用して列を明示的に除外する場合、パイプラインには、将来ソースに追加される列が自動的に含まれます。 今後の列を取り込むには、それらを一覧に追加する必要があります。詳細については、「 履歴追跡の有効化 (SCD タイプ 2)」を参照してください。
-
[ すべて実行] をクリックします。
Databricks コマンドラインインターフェース (CLI)
パイプライン定義で次のテーブル構成プロパティを使用して、取り込む特定の列を選択または選択解除できます。
-
include_columns: 必要に応じて、インジェストに含める列の一覧を指定します。 このオプションを使用して列を明示的に含める場合、パイプラインは将来ソースに追加される列を自動的に除外します。 今後の列を取り込むには、それらを一覧に追加する必要があります。 -
exclude_columns: 必要に応じて、インジェストから除外する列の一覧を指定します。 このオプションを使用して列を明示的に除外する場合、パイプラインには、将来ソースに追加される列が自動的に含まれます。 今後の列を取り込むには、それらを一覧に追加する必要があります。
パイプラインを作成するために
databricks pipelines create --json "<pipeline definition or json file path>"
パイプラインを編集するには:
databricks pipelines update --json "<pipeline definition or json file path>"
パイプライン定義を取得するには:
databricks pipelines get "<pipeline-id>"
パイプラインを削除するには:
databricks pipelines delete "<pipeline-id>"
詳細情報を得るには、以下のコマンドを実行してください。
databricks pipelines --help
databricks pipelines <create|update|get|delete|...> --help
JSON パイプライン定義の例:
"ingestion_definition": {
"connection_name": "<connection-name>",
"objects": [
{
"table": {
"source_schema": "<source-schema>",
"source_table": "<source-table>",
"destination_catalog": "<destination-catalog>",
"destination_schema": "<destination-schema>",
"table_configuration": {
"scd_type": "SCD_TYPE_2",
"include_columns": ["<column-a>", "<column-b>", "<column-c>"]
}
}
}
]
}