SharePointからファイルを取り込む

構造化ファイル、半構造化ファイル、非構造化ファイルを Microsoft SharePoint から Delta テーブルに取り込むことができます。 SQLおよびSpark APIは、Auto Loader、spark.read、COPY INTOなどのバッチおよびストリーミングAPIを用いた、Unityカタログガバナンスを利用した、増分的にSharePointファイルの取り込みをサポートしています。

SharePointからの取り込み方法を選択する

Lakeflow ConnectはSharePointから取り込む2つの方法を提供しています。 どちらもSharePointのデータにアクセスしますが、管理レベルが異なります。

主な機能

SQLおよびSpark APIは以下の機能を提供します:

• 構造化ファイル、半構造化ファイル、および非構造化ファイルの取り込み
• 詳細なインジェスト: 特定のサイト、サブサイト、ドキュメント ライブラリ、フォルダ、または単一のファイルのインジェスト
• Databricks Runtime 19 以降での SharePoint Site Pages (.aspx) の取り込み 「Ingest SharePoint サイトページ」をご覧ください。
• バッチおよびストリーミング取り込みをspark.read、オートローダー、そしてCOPY INTOを使用して行う。
• CSV やExcelなどの構造化形式と半構造化形式の自動スキーマ推論と進化
• Unity カタログ接続を使用して資格情報ストレージをセキュリティで保護する
• パターン マッチングを使用したファイルの選択 pathGlobFilter

Requirements

SharePointからファイルを取り込むには、次のものが必要です。

• Unity Catalog が有効になっているワークスペース。
• SharePoint接続を作成するための CREATE CONNECTION 特権、または クラスター アクセス モードに基づいて既存のものを使用するための適切な特権:
  • 専用アクセス モード: MANAGE CONNECTION。
  • 標準アクセス モード: USE CONNECTION。
• Databricksのランタイムバージョン17.3 LTS以上を使用するコンピュートです。
• Sites.Read.AllまたはSites.Selectedアクセス許可スコープを使用して設定された OAuth 認証。

接続を作成する

SharePoint資格情報を格納する Unity カタログ接続を作成します。 接続設定のプロセスは、SQLやSpark APIを使う場合でも、管理されたSharePointコネクタを使う場合でも同じです。

OAuth 認証オプションを含む完全な接続セットアップ手順については、「 SharePoint インジェストセットアップの概要を参照してください。

SharePoint からファイルを読み取る

ファイルを読み取るために、databricks.connection オプションを使用して作成した接続と、アクセスするSharePoint リソースを指す URL を渡します。 指定した URL によって、インジェストのスコープが決まります。

Databricks Runtime 17.3 LTS以上でサポートされているパスタイプは以下の通りです:

Databricks Runtime 18 LTS以降は、以下のパスタイプのサポートを追加しています:

Databricks Runtime 19以降は、以下のパスタイプのサポートを追加しています:

例示

SQLやSpark APIを使って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-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');

Lakeflow パイプラインSharePointファイルを取り込む

次の例では、Lakeflow パイプラインで自動ローダーを使用して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("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")
  )

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"
);

非構造化ファイルを解析する

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 パイプラインを使用した増分解析

また、Lakeflow パイプライン内で 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パイプラインは常に最新のドキュメント表現に基づいて動作します。

サポートされている形式や、出力スキーマを固定するversionオプションなどの高度なオプションについては、ai_parse_document functionを参照してください。

SharePoint メタデータ列

_sharepoint_metadata 列は、Microsoft Graph driveItem リソースから取得された、取り込まれたファイルのSharePoint固有のプロパティへのアクセスを提供する非表示のメタデータ列です。 Databricks Runtime 18 LTS 以上が必要であり、SharePointから読み取るときにすべてのファイル形式で使用できます。 返された DataFrame に _sharepoint_metadata 列を含めるには、読み取りクエリで明示的に選択する必要があります。

データ ソースに _sharepoint_metadata という名前の列が含まれている場合、SharePoint メタデータ列の名前が __sharepoint_metadata (先頭にアンダースコアが追加) に変更されて重複除去されます。 名前が一意になるまで、追加のアンダースコアが追加されます。

ファイル パスやサイズなどの一般的なファイル メタデータは、 _metadata 列を使用して照会できます。 「ファイル メタデータ列」を参照してください。

スキーマ

_sharepoint_metadata列は、次のフィールドを含むSTRUCTです。 すべてのフィールドは null にできるです。

例

次の例では、 _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サイトページをインジェストする

SharePointサイトのページを、ドキュメントライブラリとは別に、サイトのサイトのページライブラリに.aspxファイルとして保存します。 サイトページを取り込むにはDatabricks Runtime 19以上が必要です。 サイトページの詳細については、SharePointページおよびページモデルをご覧ください。

サイトページを取り込む方法は以下の通りです:

サイトページを他のファイルと同様に取り込む:

Python

# 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"))

SQL

-- 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、その他のインターフェースは、Site Pagesを他のファイルと同様に扱います。 摂取例は 「例 」を参照してください。

制限事項

SQLおよびSpark APIには以下の制限があります。

• 以下のMicrosoft 365全国クラウド展開はサポートされていません:
  • 政府コミュニティクラウド高校(GCCハイ): .sharepoint.us
  • 国防総省(DoD): .sharepoint-mil.us
  • 中国で 21Vianet が運営する Microsoft 365: .sharepoint.cn
• 同じクエリを使用して複数のサイトを取り込むことはできません。 2 つのサイトから取り込むには、2 つの個別のクエリを記述する必要があります。
• pathGlobFilter オプションを使用すると、名前でファイルをフィルター処理できます。 フォルダー パスベースのフィルター処理はサポートされていません。
• SharePointリストはサポートされていません。
• SharePoint サーバーへの書き戻しはサポートされていません。
• 自動ローダー cleanSource (インジェスト後のソースでのファイルの削除またはアーカイブ) はサポートされていません。

次のステップ

• 高度なストリーミング インジェスト パターン用 の自動ローダー について説明します
• COPY INTO について、べき等な増分読み込みを掘り下げます
• クラウド オブジェクト ストレージインジェスト パターンとの比較
• インジェスト ワークフローを自動化するように ジョブ スケジュール を設定する
• 変換処理を含むエンドツーエンドのデータ パイプラインを構築するには、Lakeflow パイプラインを使用します