Important
この機能は ベータ版です。 ワークスペース管理者は、[ プレビュー] ページからこの機能へのアクセスを制御できます。 Manage Azure Databricks プレビューを参照してください。
:::note コンプライアンス
SharePoint コネクタは、セキュリティとコンプライアンスの強化設定を構成が有効になっているワークスペースでの使用をサポートします。
:::
構造化ファイル、半構造化ファイル、非構造化ファイルを Microsoft SharePoint から Delta テーブルに取り込むことができます。 SharePoint コネクタでは、バッチ API とストリーミング API (自動ローダー、spark.read、COPY INTOなど) を使用したSharePoint ファイルの増分インジェストがサポートされます。すべて Unity カタログ ガバナンスを使用します。
SharePoint コネクタを選択します
Lakeflow Connect には、2 つの補完的なSharePoint コネクタが用意されています。 どちらもSharePoint内のデータにアクセスしますが、個別の目標をサポートします。
| 考慮事項 | マネージド SharePoint コネクタ | Standard SharePoint コネクタ |
|---|---|---|
| 管理とカスタマイズ | フル マネージド コネクタ。 Delta テーブルにデータを取り込み、ソースとの同期を維持するエンタープライズ アプリケーション向けのシンプルでメンテナンスの少ないコネクタ。 Lakeflow Connect のマネージド コネクタを参照してください。 |
バッチ API とストリーミング API ( read_files、 spark.read、 COPY INTO、自動ローダーなど) を使用して、SQL、PySpark、または Lakeflow Spark 宣言パイプラインを使用してカスタム インジェスト パイプラインを構築します。インジェスト中に複雑な変換を柔軟に実行できる一方で、パイプラインの管理と保守に対する責任が高くなります。 |
| 出力形式 | 均一バイナリ コンテンツ テーブル。 各ファイルをバイナリ形式 (1 行に 1 つのファイル) でインジェストし、ファイルメタデータも含めて、 追加の列。 |
構造化デルタ テーブル。 構造化ファイル (CSV やExcelなど) を Delta テーブルとして取り込みます。 取り込みにも使用できます バイナリ形式の非構造化ファイル。 |
| 粒度、フィルター処理、および選択 | 現在、サブフォルダーまたはファイル レベルは選択されていません。 パターン ベースのフィルター処理はありません。 指定したSharePoint ドキュメント ライブラリ内のすべてのファイルを取り込みます。 |
きめ細かくカスタム。 ドキュメント ライブラリ、サブフォルダー、または個々のファイルから取り込む URL ベースの選択。 また、 pathGlobFilter オプションを使用したパターン ベースのフィルター処理もサポートしています。 |
主な機能
標準SharePoint コネクタには次のものが用意されています。
- 構造化ファイル、半構造化ファイル、および非構造化ファイルの取り込み
- 詳細インジェスト: 特定のサイト、サブサイト、ドキュメントライブラリ、フォルダー、または1つのファイルをインジェストする
- バッチおよびストリーミング取り込みを
spark.read、オートローダー、そしてCOPY INTOを使用して行う。 - CSV やExcelなどの構造化形式と半構造化形式の自動スキーマ推論と進化
- Unity カタログ接続を使用して資格情報ストレージをセキュリティで保護する
- パターン マッチングを使用したファイルの選択
pathGlobFilter
Requirements
SharePointからファイルを取り込むには、次のものが必要です。
- Unity Catalog が有効になっているワークスペース。
- SharePoint接続を作成するための
CREATE CONNECTION特権、または クラスター アクセス モードに基づいて既存のものを使用するための適切な特権:- 専用アクセス モード:
MANAGE CONNECTION。 - 標準アクセス モード:
USE CONNECTION。
- 専用アクセス モード:
- Databricks Runtime バージョン 17.3 LTS 以降を使用するコンピューティング。
-
Sites.Read.AllまたはSites.Selectedアクセス許可スコープを使用して設定された OAuth 認証。 - SharePointベータ機能は、Previews ページから有効になります。 Manage Azure Databricks プレビューを参照してください。
- 省略可能: Excel ファイルを解析するために、Excel ベータ機能を有効にします。 「read Excel files を参照してください。
接続を作成する
SharePoint資格情報を格納する Unity カタログ接続を作成します。 接続のセットアップ プロセスは、標準コネクタとマネージド SharePoint コネクタの両方で共有されます。
OAuth 認証オプションを含む完全な接続セットアップ手順については、「 SharePoint インジェストセットアップの概要を参照してください。
SharePoint からファイルを読み取る
ファイルを読み取るために、databricks.connection オプションを使用して作成した接続と、アクセスするSharePoint リソースを指す URL を渡します。 指定した URL によって、インジェストのスコープが決まります。
Databricks Runtime 17.3 LTS 以降では、次のパスの種類がサポートされています。
| パスの種類 | Description |
|---|---|
| Site | アドレス バーからサイトの URL をコピーします。https://mytenant.sharepoint.com/sites/test-site |
| サブサイト | アドレス バーからサブサイト URL をコピーします。https://mytenant.sharepoint.com/sites/test-site/test-subsite |
| ドキュメントライブラリ |
サイト コンテンツからライブラリを開き、アドレス バーから URL をコピーします。https://mytenant.sharepoint.com/sites/test-site/Shared%20Documentshttps://mytenant.sharepoint.com/sites/test-site/custom-drive |
| Folder |
サイトのコンテンツからフォルダーを開き、アドレス バーから URL をコピーします。 または、フォルダーの Details ペインをSharePointで開き、Path の横にあるコピー アイコンをクリックします。https://mytenant.sharepoint.com/sites/test-site/Shared%20Documents/Forms/AllItems.aspx?id=%2Fsites...https://mytenant.sharepoint.com/sites/test-site/custom-drive/test-folder |
| File | ファイルを選択し、オーバーフロー メニュー (...) をクリックして、[ プレビュー] を選択します。 アドレス バーから URL をコピーします。 または、SharePointでファイルの Details ペインを開き、Path の横にあるコピー アイコンをクリックします。https://mytenant.sharepoint.com/sites/test-site/Shared%20Documents/Forms/AllItems.aspx?viewid=1a2b3c...https://mytenant.sharepoint.com/sites/test-site/custom-drive/test-folder/test.csv |
Databricks Runtime 18.3 以降では、次のパスの種類のサポートが追加されています。
| パスの種類 | Description |
|---|---|
| Tenant | アドレス バーからテナント ルート URL をコピーします。https://mytenant.sharepoint.com |
| 入れ子になったサブサイト | アドレス バーからサブサイト URL をコピーします。https://mytenant.sharepoint.com/sites/test-site/subsite/nested-subsite/nested-nested-subsite |
| リンクを共有する | ファイルまたはフォルダーを選択し、オーバーフロー メニュー (...) をクリックして、[ リンクのコピー] を選択します。 Databricks では、共有リンクを無期限に設定することをお勧めします。https://mytenant.sharepoint.com/:i:/s/test-site/1A2B3C4D5E6F7G8H9I |
| Web 用のMicrosoft 365 (旧称 Office) | web のMicrosoft 365でファイルを開き、アドレス バーから URL をコピーします。https://mytenant.sharepoint.com/:x:/r/sites/test-site/_layouts/15/Doc.aspx?sourcedoc=%1A2B... |
例示
標準のSharePoint コネクタを使用してファイルを読み取る方法はいくつかあります。
Stream SharePoint ファイルを自動ローダーでストリームする
自動ローダーは、SharePointから構造化ファイルを増分的に取り込むための最も効率的な方法を提供します。 新しいファイルが自動的に検出され、到着すると処理されます。 また、自動スキーマ推論と進化を使用して、CSV や JSON などの構造化された半構造化ファイルを取り込むこともできます。 自動ローダーの使用方法の詳細については、「 一般的なデータ読み込みパターン」を参照してください。
# Incrementally ingest new PDF files
df = (spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.option("cloudFiles.schemaLocation", <path to a schema location>)
.option("pathGlobFilter", "*.pdf")
.load("https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents")
)
# Incrementally ingest CSV files with automatic schema inference and evolution
df = (spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("databricks.connection", "my_sharepoint_conn")
.option("pathGlobFilter", "*.csv")
.option("inferColumnTypes", True)
.option("header", True)
.load("https://mytenant.sharepoint.com/sites/Engineering/Data/IoT_Logs")
)
Spark バッチ読み取りを使用したSharePoint ファイルの読み取り
次の例では、spark.read 関数を使用して、Python内のSharePoint ファイルを取り込む方法を示します。
# Read unstructured data as binary files
df = (spark.read
.format("binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.option("recursiveFileLookup", True)
.option("pathGlobFilter", "*.pdf") # optional. Example: only ingest PDFs
.load("https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents"))
# Read a batch of CSV files, infer the schema, and load the data into a DataFrame
df = (spark.read
.format("csv")
.option("databricks.connection", "my_sharepoint_conn")
.option("pathGlobFilter", "*.csv")
.option("recursiveFileLookup", True)
.option("inferSchema", True)
.option("header", True)
.load("https://mytenant.sharepoint.com/sites/Engineering/Data/IoT_Logs"))
# Read a specific Excel file from SharePoint, infer the schema, and load the data into a DataFrame
df = (spark.read
.format("excel")
.option("databricks.connection", "my_sharepoint_conn")
.option("headerRows", 1) # optional
.option("dataAddress", "Sheet1!A1:M20") # optional
.load("https://mytenant.sharepoint.com/sites/Finance/Shared%20Documents/Monthly/Report-Oct.xlsx"))
Spark SQL を使用したSharePoint ファイルの読み取り
次の例は、read_files テーブル値関数を使用して、SQL SharePoint ファイルを取り込む方法を示しています。
read_filesの使用方法の詳細については、テーブル値関数read_files参照してください。
-- Read pdf files
CREATE TABLE my_table AS
SELECT * FROM read_files(
"https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents",
`databricks.connection` => "my_sharepoint_conn",
format => "binaryFile",
pathGlobFilter => "*.pdf", -- optional. Example: only ingest PDFs
schemaEvolutionMode => "none"
);
-- Read a specific Excel sheet and range
CREATE TABLE my_sheet_table AS
SELECT * FROM read_files(
"https://mytenant.sharepoint.com/sites/Finance/Shared%20Documents/Monthly/Report-Oct.xlsx",
`databricks.connection` => "my_sharepoint_conn",
format => "excel",
headerRows => 1, -- optional
dataAddress => "Sheet1!A2:D10", -- optional
schemaEvolutionMode => "none"
);
増分インジェストCOPY INTO
COPY INTO は、Delta テーブルへのファイルのべき等増分読み込みを提供します。
COPY INTOの使用方法の詳細については、「COPY INTOを使用した一般的なデータ読み込みパターン」を参照してください。
CREATE TABLE IF NOT EXISTS sharepoint_pdf_table;
CREATE TABLE IF NOT EXISTS sharepoint_csv_table;
CREATE TABLE IF NOT EXISTS sharepoint_excel_table;
# Incrementally ingest new PDF files
COPY INTO sharepoint_pdf_table
FROM "https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents"
FILEFORMAT = BINARYFILE
PATTERN = '*.pdf'
FORMAT_OPTIONS ('databricks.connection' = 'my_sharepoint_conn')
COPY_OPTIONS ('mergeSchema' = 'true');
# Incrementally ingest CSV files with automatic schema inference and evolution
COPY INTO sharepoint_csv_table
FROM "https://mytenant.sharepoint.com/sites/Engineering/Data/IoT_Logs"
FILEFORMAT = CSV
PATTERN = '*.csv'
FORMAT_OPTIONS ('databricks.connection' = 'my_sharepoint_conn', 'header' = 'true', 'inferSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');
# Ingest a single Excel file
COPY INTO sharepoint_excel_table
FROM "https://mytenant.sharepoint.com/sites/Finance/Shared%20Documents/Monthly/Report-Oct.xlsx"
FILEFORMAT = EXCEL
FORMAT_OPTIONS ('databricks.connection' = 'my_sharepoint_conn', 'headerRows' = '1')
COPY_OPTIONS ('mergeSchema' = 'true');
Lakeflow Spark 宣言型パイプラインで SharePoint ファイルを使用する
注
SharePoint コネクタには、Databricks Runtime 17.3 以降が必要です。 コネクタを使用するには、パイプライン設定で "CHANNEL" = "PREVIEW" を設定します。 プレビューの詳細については、「 パイプラインプロパティリファレンス」を参照してください。
次の例では、Lakeflow Spark 宣言パイプラインで自動ローダーを使用してSharePointファイルを読み取る方法を示します。
Python
from pyspark import pipelines as dp
# Incrementally ingest new PDF files
@dp.table
def sharepoint_pdf_table():
return (spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.option("pathGlobFilter", "*.pdf")
.load("https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents")
)
# Incrementally ingest CSV files with automatic schema inference and evolution
@dp.table
def sharepoint_csv_table():
return (spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("databricks.connection", "my_sharepoint_conn")
.option("pathGlobFilter", "*.csv")
.option("inferColumnTypes", True)
.option("header", True)
.load("https://mytenant.sharepoint.com/sites/Engineering/Data/IoT_Logs")
)
# Read a specific Excel file from SharePoint in a materialized view
@dp.table
def sharepoint_excel_table():
return (spark.read.format("excel")
.option("databricks.connection", "my_sharepoint_conn")
.option("headerRows", 1) # optional
.option("inferColumnTypes", True) # optional
.option("dataAddress", "Sheet1!A1:M20") # optional
.load("https://mytenant.sharepoint.com/sites/Finance/Shared%20Documents/Monthly/Report-Oct.xlsx")
SQL
-- Incrementally ingest new PDF files
CREATE OR REFRESH STREAMING TABLE sharepoint_pdf_table
AS SELECT * FROM STREAM read_files(
"https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents",
format => "binaryFile",
`databricks.connection` => "my_sharepoint_conn",
pathGlobFilter => "*.pdf");
-- Incrementally ingest CSV files with automatic schema inference and evolution
CREATE OR REFRESH STREAMING TABLE sharepoint_csv_table
AS SELECT * FROM STREAM read_files(
"https://mytenant.sharepoint.com/sites/Engineering/Data/IoT_Logs",
format => "csv",
`databricks.connection` => "my_sharepoint_conn",
pathGlobFilter => "*.csv",
"header", "true");
-- Read a specific Excel file from SharePoint in a materialized view
CREATE OR REFRESH MATERIALIZED VIEW sharepoint_excel_table
AS SELECT * FROM read_files(
"https://mytenant.sharepoint.com/sites/Finance/Shared%20Documents/Monthly/Report-Oct.xlsx",
`databricks.connection` => "my_sharepoint_conn",
format => "excel",
headerRows => 1, -- optional
dataAddress => "Sheet1!A2:D10", -- optional
`cloudFiles.schemaEvolutionMode` => "none"
);
非構造化ファイルを解析する
標準の SharePoint コネクタと binaryFile 形式を使用して、SharePoint (PDF、Word ドキュメント、PowerPoint ファイルなど) から非構造化ファイルを取り込む場合、ファイルの内容は生のバイナリ データとして格納されます。 RAG、検索、分類、ドキュメントの理解などの AI ワークロード用にこれらのファイルを準備するには、 ai_parse_documentを使用して、バイナリ コンテンツを構造化されたクエリ可能な出力に解析できます。
次の例は、 documents という名前のブロンズ Delta テーブルに格納されている非構造化ドキュメントを解析し、解析されたコンテンツを含む新しい列を追加する方法を示しています。
CREATE TABLE documents AS
SELECT * FROM read_files(
"https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents",
`databricks.connection` => "my_sharepoint_conn",
format => "binaryFile",
pathGlobFilter => "*.{pdf,docx}",
schemaEvolutionMode => "none"
);
SELECT *, ai_parse_document(content) AS parsed_content
FROM documents;
parsed_content列には、抽出されたテキスト、テーブル、レイアウト情報、およびダウンストリーム AI パイプラインに直接使用できるメタデータが含まれています。
Lakeflow Spark 宣言パイプラインを使用した増分解析
また、Lakeflow Spark 宣言パイプライン内の ai_parse_document を使用して、増分解析を有効にすることもできます。 新しいファイルがSharePointから流れ込むと、パイプラインの更新時に自動的に解析されます。
たとえば、新しく取り込まれたドキュメントを継続的に解析する具体化されたビューを定義できます。
CREATE OR REFRESH STREAMING TABLE sharepoint_documents_table
AS SELECT * FROM STREAM read_files(
"https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents",
format => "binaryFile",
`databricks.connection` => "my_sharepoint_conn",
pathGlobFilter => "*.{pdf,docx}");
CREATE OR REFRESH MATERIALIZED VIEW documents_parsed
AS
SELECT *, ai_parse_document(content) AS parsed_content
FROM sharepoint_documents_table;
この方法により、次のことが保証されます。
- 新しく取り込まれたSharePoint ファイルは、具体化されたビューが更新されるたびに自動的に解析されます
- 解析された出力と受信データの同期が維持される
- ダウンストリーム AI パイプラインは、常に最新のドキュメントの表現で動作します。
詳細情報: サポートされている形式と詳細オプションについては、 ai_parse_document を参照してください。
SharePoint メタデータ列
Important
この機能は、プライベート プレビュー段階にあります。 試すには、Azure Databricks の連絡先にお問い合わせください。
_sharepoint_metadata 列は、Microsoft Graph driveItem リソースから取得された、取り込まれたファイルのSharePoint固有のプロパティへのアクセスを提供する非表示のメタデータ列です。 Databricks Runtime 18.1 以降が必要であり、SharePointから読み取るときにすべてのファイル形式で使用できます。 返された DataFrame に _sharepoint_metadata 列を含めるには、読み取りクエリで明示的に選択する必要があります。
データ ソースに _sharepoint_metadata という名前の列が含まれている場合、SharePoint メタデータ列の名前が __sharepoint_metadata (先頭にアンダースコアが追加) に変更されて重複除去されます。 名前が一意になるまで、追加のアンダースコアが追加されます。
ファイル パスやサイズなどの一般的なファイル メタデータは、 _metadata 列を使用して照会できます。 詳細については、「ファイル メタデータ列」を参照してください。
スキーマ
_sharepoint_metadata列は、次のフィールドを含むSTRUCTです。 すべてのフィールドは null にできるです。
| 名前 | タイプ | Description | 例 | Databricks の最低ランタイム リリース |
|---|---|---|---|---|
| item_id | STRING |
項目の driveItem ID。 | 01OMQ3MNLH42C5J675CBEI5CRK7SPKQUTZ |
18.1 |
| site_id | STRING |
アイテムを含むSharePoint サイトの ID。 | mytenant.sharepoint.com,69dc7b12-f92c-498d-9514-596b793a1f77,c6c1db8d-2b8d-48a1-a549-394b63d74725 |
18.1 |
| drive_id | STRING |
項目を含むドライブの ID。 | b!EnvcaSz5jUmVFFlreTofd43bwcaNK6FIpUk5S2PXRyWTvQraaWQkSpwQEgThHDS- |
18.1 |
| ドライブタイプ | STRING |
SharePoint ライブラリの場合は documentLibrary、OneDrive for Business の場合は business などのドライブの種類。 |
documentLibrary |
18.1 |
| parent_id | STRING |
親フォルダーの driveItem ID。 | 01OMQ3MNN6Y2GOVW7725BZO354PWSELRRZ |
18.1 |
| parent_name | STRING |
親フォルダーの名前。 | Shared Documents |
18.1 |
| parent_path | STRING |
親フォルダーのドライブ相対パス。 | /drives/b!EnvcaSz5.../root: |
18.1 |
| web_url | STRING |
SharePoint上のアイテムのブラウザー URL。 | https://mytenant.sharepoint.com/sites/TestSite/_layouts/15/Doc.aspx?sourcedoc=... |
18.1 |
| mime_type | STRING |
アイテムの MIME の種類。 | application/vnd.ms-excel |
18.1 |
| created_by_email | STRING |
アイテムを作成したユーザーの電子メール。 | alice@example.onmicrosoft.com |
18.1 |
| created_by_name | STRING |
アイテムを作成したユーザーの表示名。 | Alice Example |
18.1 |
| created_timestamp | TIMESTAMP |
アイテムが作成された時刻。 | 2025-12-03 13:33:12 |
18.1 |
| メールで最後に変更された人 | STRING |
アイテムを最後に変更したユーザーの電子メール。 | alice@example.onmicrosoft.com |
18.1 |
| 最終更新者名 | STRING |
アイテムを最後に変更したユーザーの表示名。 | Alice Example |
18.1 |
| etag | STRING |
アイテムのETag。 項目またはそのメタデータのいずれかが変更されたときに変更されます。 | "{D485E667-FDFB-4810-8E8A-2AFC9EA85279},1" |
18.1 |
| ctag | STRING |
アイテムの変更タグ。 アイテムのコンテンツが変更された場合にのみ変更されます。 | "c:{D485E667-FDFB-4810-8E8A-2AFC9EA85279},1" |
18.1 |
| 説明 | STRING |
項目の説明 (設定されている場合)。 | Q4 financial report |
18.1 |
| additional_metadata | VARIANT |
その他の driveItem フィールドは、Microsoft Graphによって返されますが、上記では抽出されません。 | {"shared":{"scope":"users"},...} |
18.1 |
注
additional_metadata フィールドはVARIANTとして返されます。 「 VARIANT 型」を参照してください。
例
次の例では、 _sharepoint_metadata 列を読み取りクエリに含める方法、列から特定のフィールドを選択する方法、 additional_metadataVARIANT フィールドから値を抽出する方法を示します。
Python
df = (spark.read
.format("binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.load("https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents")
.select("*", "_metadata", "_sharepoint_metadata"))
SQL
SELECT *, _sharepoint_metadata
FROM read_files(
"https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents",
`databricks.connection` => "my_sharepoint_conn",
format => "binaryFile"
);
_sharepoint_metadata構造体から特定のフィールドを選択します。
df = (spark.read
.format("binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.load("https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents")
.select("_sharepoint_metadata.item_id", "_sharepoint_metadata.etag"))
additional_metadata キャスト演算子を使用して、VARIANT:: フィールドから値を抽出します。
SELECT
*,
_sharepoint_metadata.additional_metadata:shared:scope::STRING AS shared_scope
FROM read_files(
"https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents",
`databricks.connection` => "my_sharepoint_conn",
format => "binaryFile"
);
制限事項
標準SharePoint コネクタには、次の制限があります。
- マルチサイト インジェストなし: 同じクエリを使用して複数のサイトを取り込むことはできません。 2 つのサイトから取り込むには、2 つの個別のクエリを記述する必要があります。
-
フィルター処理:
pathGlobFilterオプションを使用して、名前でファイルをフィルター処理できます。 フォルダー パスベースのフィルター処理はサポートされていません。 - サポートされていない形式: SharePoint リストと .aspx サイトページはサポートされていません。 ドキュメント ライブラリ内のファイルのみがサポートされています。
- SharePoint サーバーへの書き戻しはサポートされていません。
- 自動ローダー
cleanSource(インジェスト後のソースでのファイルの削除またはアーカイブ) はサポートされていません。
次のステップ
- 高度なストリーミング インジェスト パターン用 の自動ローダー について説明します
- COPY INTO のべき等な増分読み込みを研究する
- クラウド オブジェクト ストレージインジェスト パターンとの比較
- インジェスト ワークフローを自動化するように ジョブ スケジュール を設定する
- Lakeflow Spark 宣言型パイプラインを使用して、変換を使用してエンドツーエンドのデータ パイプラインを構築する