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

:::note コンプライアンス

SharePoint コネクタは、セキュリティとコンプライアンスの強化設定を構成が有効になっているワークスペースでの使用をサポートします。

:::

構造化ファイル、半構造化ファイル、非構造化ファイルを Microsoft SharePoint から Delta テーブルに取り込むことができます。 SharePoint コネクタでは、バッチ API とストリーミング API (自動ローダー、spark.read、COPY INTOなど) を使用したSharePoint ファイルの増分インジェストがサポートされます。すべて Unity カタログ ガバナンスを使用します。

SharePoint コネクタを選択します

Lakeflow Connect には、2 つのSharePoint コネクタが用意されています。 どちらもSharePointのデータにアクセスしますが、管理レベルが異なります。

主な機能

標準SharePoint コネクタには次のものが用意されています。

• 構造化ファイル、半構造化ファイル、および非構造化ファイルの取り込み
• 詳細なインジェスト: 特定のサイト、サブサイト、ドキュメント ライブラリ、フォルダ、または単一のファイルのインジェスト
• 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 認証。
• 省略可能: Excel ファイルを解析するために、Excel ベータ機能を有効にします。 「Excel ファイルの読み取りとストリーム配信」を参照してください。

接続を作成する

SharePoint資格情報を格納する Unity カタログ接続を作成します。 接続のセットアップ プロセスは、標準コネクタとマネージド SharePoint コネクタの両方で共有されます。

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

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

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

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

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

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

例示

標準の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"
);

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

標準の 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, 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を他のファイルと同様に扱います。 摂取例は 「例 」を参照してください。

制限事項

標準SharePoint コネクタには、次の制限があります。

• 同じクエリを使用して複数のサイトを取り込むことはできません。 2 つのサイトから取り込むには、2 つの個別のクエリを記述する必要があります。
• pathGlobFilter オプションを使用すると、名前でファイルをフィルター処理できます。 フォルダー パスベースのフィルター処理はサポートされていません。
• SharePointリストはサポートされていません。
• SharePoint サーバーへの書き戻しはサポートされていません。
• 自動ローダー cleanSource (インジェスト後のソースでのファイルの削除またはアーカイブ) はサポートされていません。

次のステップ

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