SharePointからファイルを取り込む
SharePoint からの取り込みは、強化されたセキュリティとコンプライアンスの設定を構成するが有効になっているワークスペースでの使用をサポートしています。
Microsoft SharePointから構造化ファイル、半構造化ファイル、および非構造化ファイルをDeltaテーブルに取り込むことができます。SQLおよび Spark APIs は、Auto Loader、spark.read、COPY INTO などのバッチおよびストリーミング API を使用した SharePoint ファイルの増分取り込みをサポートしており、これらはすべて Unity Catalog ガバナンスによって管理されます。
SharePoint からの取り込み方法を選択する
Lakeflow Connectには、SharePointから取り込むための2つの方法が用意されています。どちらもSharePointのデータにアクセスしますが、管理レベルが異なります。
オプション | 説明 |
|---|---|
Delta テーブルにデータを取り込み、ソースと同期を維持する、エンタープライズアプリケーション向けのフルマネージド型コネクタ。 | |
SQL および Spark API | SQL、PySpark、または LakeFlow Pipelines を使用して、 |
Databricksでは、ほとんどのユースケースでマネージド SharePoint コネクタを使用することをお勧めします。自動的な増分取り込み、幅広い形式のサポート、および柔軟なファイルとフォルダーの選択機能を備えた、フルマネージドのエクスペリエンスを提供します。
主な特徴
SQL および Spark API で提供される機能:
- 構造化ファイル、半構造化ファイル、非構造化ファイルの取り込み
- きめ細かな取り込み: 特定のサイト、サブサイト、ドキュメント ライブラリ、フォルダー、または単一のファイルを取り込む
- Databricks Runtime 19以降でのSharePointサイトページ (
.aspx) の取り込み。SharePointサイトページの取り込みを参照してください。 spark.read、 Auto Loader 、およびCOPY INTO- CSV や Excel などの構造化および半構造化形式の自動スキーマ推論と進化
- Unity Catalog接続による安全な資格情報ストレージ
- パターンマッチングによるファイル選択
pathGlobFilter
要件
SharePoint からファイルを取り込むには、次のものが必要です。
-
Unity Catalog が有効になっているワークスペース。
-
CREATE CONNECTIONSharePoint接続を作成するための権限、またはクラスターアクセスモードに基づいて既存の接続を使用するための適切な権限。- 専用アクセスMode :
MANAGE CONNECTION。 - 標準アクセスMode :
USE CONNECTION。
- 専用アクセスMode :
-
Databricks Runtimeバージョン17.3 LTS以降を使用するコンピュート。
-
OAuth 認証は
Sites.Read.AllまたはSites.Selected権限スコープで設定されています。
接続を作成する
SharePoint の認証情報を保存するための Unity Catalog 接続を作成します。SQL および Spark API を使用する場合でも、管理対象の SharePoint コネクタを使用する場合でも、接続のセットアッププロセスは同じです。
OAuth 認証オプションを含む完全な接続セットアップ手順については、 「SharePoint 取り込みセットアップの概要」を参照してください。
SharePointからファイルを読み取る
ファイルを読み取るには、 databricks.connectionオプションを使用して作成した接続と、アクセスしたい SharePoint リソースを指す URL を渡します。ご提供いただくURLによって、データ取り込みの範囲が決まります。
Databricks Runtime 17.3 LTS以降では、以下のパスタイプがサポートされています。
パスの種類 | 説明 |
|---|---|
サイト | アドレスバーからサイトのURLをコピーしてください。
|
サブサイト | アドレスバーからサブサイトのURLをコピーしてください。
|
ドキュメントライブラリ | サイトコンテンツ からライブラリを開き、アドレスバーからURLをコピーしてください。
|
フォルダ | サイトコンテンツ からフォルダを開き、アドレスバーからURLをコピーします。または、SharePoint でフォルダーの 詳細 ペインを開き、 [パス] の横にあるコピーアイコンをクリックします。
|
ファイル | ファイルを選択し、オーバーフローメニュー( … )をクリックして、 プレビュー を選択します。アドレスバーからURLをコピーしてください。または、SharePoint でファイルの 詳細 ペインを開き、 [パス] の横にあるコピーアイコンをクリックします。
|
Databricks Runtime 18 LTS 以降では、次のパスタイプがサポートされます。
パスの種類 | 説明 |
|---|---|
ネストされたサブサイト | アドレスバーからサブサイトのURLをコピーしてください。
|
共有リンク | ファイルまたはフォルダを選択し、オーバーフローメニュー( ... )をクリックして、 [リンクをコピー] を選択します。Databricksは、共有リンクの有効期限を無期限に設定することを推奨しています。
|
Microsoft 365 for the web(旧称:Office) | Microsoft 365 for the webでファイルを開き、アドレスバーからURLをコピーしてください。
|
Databricks Runtime 19以降では、以下のパスタイプがサポートされます:
パスの種類 | 説明 |
|---|---|
テナント | アドレスバーからテナントのルートURLをコピーしてください。 テナント全体のインジェストを実行するには、OAuthマシン間(M2M)接続を使用する必要があります。Microsoft Graphの
|
サイトページ | ドキュメントライブラリの外部に保存されている、サイトのサイトページ (
|
例
SQL および Spark API を使用して SharePoint からファイルを読み込む方法には、いくつかの種類があります。
Auto Loaderを使用して SharePoint ファイルをストリーミングする
Auto Loader は、SharePoint から構造化ファイルを段階的に取り込む最も効率的な方法を提供します。新しいファイルが自動的に検出され、到着すると処理されます。また、自動スキーマ推論および進化により、CSV や JSON などの構造化ファイルや半構造化ファイルを取り込むこともできます。Auto Loader使用法の詳細については、 「一般的なデータ読み込みパターン」を参照してください。
# 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-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("cloudFiles.schemaLocation", "<path-to-schema-location>")
.option("pathGlobFilter", "*.csv")
.option("cloudFiles.inferColumnTypes", True)
.option("header", True)
.load("https://mytenant.sharepoint.com/sites/Engineering/Data/IoT_Logs")
)
Spark バッチ読み取りを使用して SharePoint ファイルを読み取る
次の例では、spark.read関数を使用してPythonでSharePointファイルを取り込む方法を示します。サイト内のドキュメントライブラリ、ドキュメントライブラリ内のフォルダ、またはサイト内のサブサイトなど、ネストされた構造からファイルを読み取るには、recursiveFileLookupオプションをtrueに設定します。
# 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');
Ingest SharePoint files in LakeFlow Pipelines
SQL および Spark API を使用した SharePoint ファイルの取り込みには、Databricks Runtime 17.3 以降が必要です。
次の例は、LakeFlow PipelinesでAuto Loaderを使用してSharePointファイルを読み取る方法を示しています。
- Python
- SQL
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("cloudFiles.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")
)
-- 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"
);
非構造化ファイルを解析する
SQL および Spark API と 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, map('version', '2.0')) AS parsed_content
FROM documents;
parsed_content列には、下流の AI パイプラインに直接使用できる抽出されたテキスト、テーブル、レイアウト情報、メタデータが含まれています。
LakeFlow Pipelines による増分解析
増分解析を有効にするために、Lakeflow pipelines内でai_parse_documentを使用することもできます。SharePointから新しいファイルがストリームとして入ってくると、パイプラインが更新される際に自動的に解析されます。
例えば、未加工のバイナリファイルを取り込むブロンズストリーミングテーブルを定義し、次に解析されたコンテンツを新しい列に格納する2つ目のストリーミングテーブルを定義できます:
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 STREAMING TABLE documents_parsed
AS SELECT
*,
ai_parse_document(content, map('version', '2.0')) AS parsed_content
FROM STREAM sharepoint_documents_table;
このアプローチにより、次のことが保証されます。
- 新しく取り込まれたSharePointファイルは、パイプラインが更新されるたびに自動的に解析されます。
- 解析された出力は、受信データと同期した状態を維持します。
- 下流のAIパイプラインは、常に最新のドキュメント表現で動作します。
サポートされている形式と詳細オプションについては、ai_parse_document関数を参照してください。これには、出力スキーマをピン留めするversionオプションが含まれます。
SharePoint メタデータ列
_sharepoint_metadata列は、取り込まれたファイルのSharePoint固有のプロパティへのアクセスを提供する非表示のメタデータ列であり、Microsoft Graph driveItemリソースから取得されます。これにはDatabricks Runtime 18 LTS以上が必要であり、SharePointから読み取るときにすべてのファイル形式で利用できます。返されたDataFrameに_sharepoint_metadata列を含めるには、読み取りクエリーで明示的に選択する必要があります。
データソースに_sharepoint_metadataという名前の列が含まれている場合、重複排除のため、SharePointメタデータ列の名前は__sharepoint_metadata (先頭にアンダースコアを追加)に変更されます。名前が一意になるまで、アンダースコアが追加されます。
ファイル パスやサイズなどの一般的なファイル メタデータは、_metadata 列を使用してクエリーできます。ファイルのメタデータ列を参照してください。
スキーマ
_sharepoint_metadata列は、以下のフィールドを含むSTRUCTです。すべてのフィールドはnullを許容します。
名前 | Type | 説明 | 例 |
|---|---|---|---|
アイテムID |
| アイテムのドライブアイテムID。 |
|
サイトID |
| アイテムが含まれているSharePointサイトのID。 |
|
ドライブID |
| アイテムが格納されているドライブのID。 |
|
ドライブタイプ |
| ドライブの種類。たとえば、SharePoint ライブラリの場合は |
|
親ID |
| 親フォルダーのドライブアイテムID。 |
|
親の名前 |
| 親フォルダの名前。 |
|
親パス |
| 親フォルダのドライブ相対パス。 |
|
ウェブURL |
| SharePoint上のアイテムのブラウザURL。 |
|
mime_type |
| アイテムのMIMEタイプ。 |
|
電子メールで作成されました |
| アイテムを作成したユーザーのメールアドレス。 |
|
作成者名 |
| アイテムを作成したユーザーの表示名。 |
|
作成日時 |
| そのアイテムが作成された時刻。 |
|
最終更新者(メール) |
| 最後にアイテムを編集したユーザーのメールアドレス。 |
|
最終更新者名 |
| 最後にアイテムを変更したユーザーの表示名。 |
|
etag |
| 商品のEタグ。アイテムまたはそのメタデータのいずれかが変更された場合に変更されます。 |
|
ctag |
| 商品の変更タグ。アイテムの内容が変更された場合にのみ変更されます。 |
|
説明 |
| 設定されている場合は、商品の説明が表示されます。 |
|
追加のメタデータ |
| Microsoft Graphによって返されるが、上記で抽出されなかったその他のdriveItemフィールド。 |
|
additional_metadataフィールドはVARIANTとして返されます。VARIANT型を参照してください。
例
以下の例は、読み取りクエリに_sharepoint_metadata列を含める方法、列から特定のフィールドを選択する方法、およびadditional_metadata VARIANTフィールドから値を抽出する方法を示しています。
- Python
- SQL
df = (spark.read
.format("binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.load("https://mytenant.sharepoint.com/sites/Marketing/Shared%20Documents")
.select("*", "_metadata", "_sharepoint_metadata"))
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サイトページの取り込み
SharePointは、サイトの Site Pages ライブラリにSite Pagesを.aspxファイルとして保存します。これはドキュメントライブラリとは別個のものです。Site Pagesの取り込みには、Databricks Runtime 19以上が必要です。Site Pagesの詳細については、「SharePoint pages and the page model」を参照してください。
サイトページは、以下の方法でインジェストできます。
パスの種類 | 説明 |
|---|---|
単一ページ | SharePointでページを開き、アドレスバーからURLをコピーします。
|
Site Pagesライブラリ | サイトコンテンツ から サイトページ を開き、アドレスバーからURLをコピーします。
|
サイトまたはサブサイト | サイトまたはサブサイトを取り込む際、Databricksはサイトのドキュメントライブラリと併せてSite Pagesライブラリを取り込みます。
|
他のファイルと同様にサイトページを取り込む:
- Python
- SQL
# Read a single Site Page
df = (spark.read
.format("binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.load("https://mytenant.sharepoint.com/sites/test-site/SitePages/my-news-post.aspx"))
# Read a site's Site Pages library
df = (spark.read
.format("binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.option("pathGlobFilter", "*.aspx") # optional. Example: only ingest Site Page files
.load("https://mytenant.sharepoint.com/sites/test-site/SitePages"))
# Read all Site Pages from a site
df = (spark.read
.format("binaryFile")
.option("databricks.connection", "my_sharepoint_conn")
.option("recursiveFileLookup", True)
.option("pathGlobFilter", "*.aspx") # optional. Example: only ingest Site Page files
.load("https://mytenant.sharepoint.com/sites/test-site"))
-- Read a single Site Page
CREATE TABLE site_page AS
SELECT * FROM read_files(
"https://mytenant.sharepoint.com/sites/test-site/SitePages/my-news-post.aspx",
`databricks.connection` => "my_sharepoint_conn",
format => "binaryFile"
);
-- Read a site's Site Pages library
CREATE TABLE site_pages AS
SELECT * FROM read_files(
"https://mytenant.sharepoint.com/sites/test-site/SitePages",
`databricks.connection` => "my_sharepoint_conn",
format => "binaryFile",
pathGlobFilter => "*.aspx" -- optional. Example: only ingest Site Page files
);
-- Read all Site Pages from a site
CREATE TABLE site_with_pages AS
SELECT * FROM read_files(
"https://mytenant.sharepoint.com/sites/test-site",
`databricks.connection` => "my_sharepoint_conn",
format => "binaryFile",
recursiveFileLookup => true,
pathGlobFilter => "*.aspx" -- optional. Example: only ingest Site Page files
);
Auto Loader、COPY INTO、およびその他のインターフェースは、他のファイルと同様にサイトページで動作します。取り込みの例については、例を参照してください。
制限事項
SQL および Spark API には以下の制限があります。
-
次の Microsoft 365 ナショナルクラウドデプロイメントはサポートされていません。
- Government Community Cloud High (GCC High):
.sharepoint.us - 国防総省(DoD):
.sharepoint-mil.us - 中国の21Vianetが運営するMicrosoft 365:
.sharepoint.cn
- Government Community Cloud High (GCC High):
-
同じクエリで複数のサイトを取り込むことはできません。2つのサイトから取り込むには、2つの個別のクエリを作成する必要があります。
-
pathGlobFilterオプションを使用して、ファイル名をフィルタリングできます。フォルダパスによるフィルタリングはサポートされていません。 -
SharePointリストはサポートされていません。
-
SharePoint サーバーへの書き戻しはサポートされていません。
-
Auto Loader
cleanSource(取り込み後にソースでファイルを削除またはアーカイブする) はサポートされていません。
次のステップ
- 高度なストリーミング取り込みパターンのためのAuto Loaderについて学ぶ
- べき等増分ロードのCOPY INTOを調べる
- クラウドオブジェクトストレージの取り込みパターンと比較する
- ジョブスケジュールを設定して取り込みワークフローを自動化する
- 変換を伴うエンドツーエンドのデータパイプラインを構築するには、LakeFlow Pipelinesを使用します。