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

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページから Lakeflow Connect for OneDrive For Business プレビューにオプトインすることで、これを有効にできます。Databricksのプレビューを管理するを参照してください。

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

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

:::

ユーザーのMicrosoft OneDrive for Business（個人用の マイファイル ドライブ）から、構造化データ、半構造化データ、非構造化ファイルをDeltaテーブルに取り込むことができます。標準のOneDriveコネクタは、Auto Loader、spark.read、COPY INTOを含むバッチおよびストリーミングAPIを使用したOneDriveファイルの増分取り込みをサポートしており、すべてUnity Catalogガバナンスが適用されます。これらのAPIを使用して、OneDrive内のファイルから直接Spark DataFrames、マテリアライズドビュー、ストリーミングテーブルを作成します。

OneDriveコネクタは、個々のユーザーの個人用ドライブからファイルを取り込みます。SharePointサイト、チームサイト、または共有ドキュメントライブラリから取り込むには、代わりに SharePointコネクタ を使用してください。

主な機能

標準の OneDrive コネクタは以下を提供します：

• ユーザーの個人用 OneDrive からの構造化データ、半構造化データ、および非構造化ファイルの取り込み。
• きめ細かな取り込み: ドライブ全体、フォルダー、または単一のファイルを取り込みます。
• spark.read、Auto Loader、および COPY INTO を使用したバッチおよびストリーミングの取り込み。
• CSVなどの構造化および半構造化フォーマットに対する自動スキーマ推論と進化。
• Unity Catalog接続を使用して認証情報を安全に保存します。
• pathGlobFilter を使用したパターンマッチングによるファイル選択。

要件

OneDriveからファイルを取り込むには、以下が必要です：

• Unity Catalog が有効になっているワークスペース。

• CREATE CONNECTION OneDrive接続を作成するための権限、またはクラスターアクセスモードに基づいて既存の接続を使用するための適切な権限：

  • Dedicated Access Mode: MANAGE CONNECTION。
  • 標準アクセスモード：USE CONNECTION。

• Databricks Runtimeバージョン19以降を使用するコンピュート。

• Files.Read.All権限スコープで設定されたOAuth認証。

接続を作成

OneDriveの認証情報を保存するために、Unity Catalog接続を作成します。OneDriveコネクタは、Microsoft Entra ID（Azure AD）およびMicrosoft Graphに対してOAuthユーザー・ツー・マシン（U2M）認証を使用します。

1. Databricksワークスペースで、 カタログ > 外部ロケーション > 接続 > 接続の作成 を選択します。
2. 接続の設定 ウィザードの 接続の基本 ページで、一意の 接続名 を指定します。
3. 「接続の種類」 ドロップダウンメニューで、 OneDrive を検索して選択します。
4. （オプション）コメントを追加します。
5. [次へ] を選択します。
6. [認証] ページで、読み取り対象の OneDrive を持つ Microsoft アカウントでサインインし、要求された Microsoft Graph 権限 (Files.Read.All および offline_access) に同意します。
7. Databricksワークスペースにリダイレクトされたら、[ Create connection ] を選択します。

この接続はサインインしたユーザーとしてファイルを読み取るため、コネクタはユーザーが OneDrive でアクセスできるファイルにのみアクセスできます。

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

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

次のパスタイプがサポートされています：

パスタイプ	説明
ドライブ全体	アドレスバーからOneDriveのルートURLをコピーします。 https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents
フォルダ	OneDrive でフォルダーを開き、アドレスバーから URL をコピーします。 https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/_layouts/15/onedrive.aspx?id=%2Fpersonal%2Fuser%2FDocuments%2FReports
ファイル	ファイルを選択し、オーバーフローメニュー（ ... ）をクリックして、 ブラウザで開く または 詳細 を選択してから、URLをコピーします。 https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/Reports/report.csv
Linkを共有	ファイルまたはフォルダを選択し、 共有 または Copy Link をクリックして、生成されたリンクを使用します。Databricksでは、共有Linkを無期限に設定することを推奨しています。 https://mytenant-my.sharepoint.com/:x:/g/personal/user_mytenant_onmicrosoft_com/Eab12...

パスタイプ	説明
ドライブ全体	アドレスバーからOneDriveのルートURLをコピーします。 https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents
フォルダ	OneDrive でフォルダーを開き、アドレスバーから URL をコピーします。 https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/_layouts/15/onedrive.aspx?id=%2Fpersonal%2Fuser%2FDocuments%2FReports
ファイル	ファイルを選択し、オーバーフローメニュー（ ... ）をクリックして、 ブラウザで開く または 詳細 を選択してから、URLをコピーします。 https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/Reports/report.csv
Linkを共有	ファイルまたはフォルダを選択し、 共有 または Copy Link をクリックして、生成されたリンクを使用します。Databricksでは、共有Linkを無期限に設定することを推奨しています。 https://mytenant-my.sharepoint.com/:x:/g/personal/user_mytenant_onmicrosoft_com/Eab12...

注記

OneDriveのURLは個人用の-myホスト（例：mytenant-my.sharepoint.com）を使用しますが、これはSharePointホスト（mytenant.sharepoint.com）とは異なります。SharePointサイトへの共有LinkはOneDriveコネクタでは受け付けられません。それらにはSharePointコネクタを使用してください。

URLは、個人のドライブ、その中のフォルダーやファイル、または共有Linkに解決される必要があります。ベアホストURL（例: https://mytenant-my.sharepoint.com/）や、ドライブを指していないパス（例: https://mytenant-my.sharepoint.com/shared）はサポートされていません。

例

標準の OneDrive コネクタを使用してファイルを読み取る方法はいくつかあります。

Auto Loader を使用して OneDrive ファイルをストリームする

Auto Loaderは、OneDriveからファイルを増分取り込みするための最も効率的な方法を提供します。新しいファイルを自動的に検出し、到着時に処理します。また、自動スキーマ推論と進化機能を使用して、CSVやJSONなどの構造化ファイルおよび半構造化ファイルを取り込むこともできます。Auto Loaderの使用方法の詳細については、「一般的なデータ読み込みパターン」を参照してください。

Python

# Incrementally ingest new PDF files
df = (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "binaryFile")
    .option("databricks.connection", "my_onedrive_conn")
    .option("cloudFiles.schemaLocation", <path to a schema location>)
    .option("pathGlobFilter", "*.pdf")
    .load("https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents")
)

# Incrementally ingest CSV files with automatic schema inference and evolution
df = (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("databricks.connection", "my_onedrive_conn")
    .option("cloudFiles.schemaLocation", <path to a schema location>)
    .option("pathGlobFilter", "*.csv")
    .option("cloudFiles.inferColumnTypes", True)
    .option("header", True)
    .load("https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/IoT_Logs")
)

Sparkバッチ読み取りを使用したOneDriveファイルの読み取り

次の例では、spark.read関数を使用してPythonでOneDriveファイルを取り込む方法を示します。OneDriveのネストされたフォルダからファイルを読み取るには、recursiveFileLookupオプションをtrueに設定します。

Python

# Read unstructured data as binary files
df = (spark.read
        .format("binaryFile")
        .option("databricks.connection", "my_onedrive_conn")
        .option("recursiveFileLookup", True)
        .option("pathGlobFilter", "*.pdf") # optional. Example: only ingest PDFs
        .load("https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents"))

# 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_onedrive_conn")
        .option("pathGlobFilter", "*.csv")
        .option("recursiveFileLookup", True)
        .option("inferSchema", True)
        .option("header", True)
        .load("https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/IoT_Logs"))

Spark SQLを使用してOneDriveファイルを読み取る

次の例は、read_files テーブル値関数を使用して SQL で OneDrive ファイルを取り込む方法を示しています。read_files の使用方法の詳細については、read_files テーブル値関数を参照してください。

SQL

-- Read pdf files
CREATE TABLE my_table AS
SELECT * FROM read_files(
  "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents",
  `databricks.connection` => "my_onedrive_conn",
  format => "binaryFile",
  pathGlobFilter => "*.pdf", -- optional. Example: only ingest PDFs
  schemaEvolutionMode => "none"
);

-- Read CSV files with automatic schema inference
CREATE TABLE my_csv_table AS
SELECT * FROM read_files(
  "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/IoT_Logs",
  `databricks.connection` => "my_onedrive_conn",
  format => "csv",
  header => true,
  schemaEvolutionMode => "none"
);

Ingest OneDrive files in LakeFlow Pipelines

注記

OneDrive コネクタには Databricks Runtime 19以降が必要です。コネクタを使用するには、パイプライン設定で "CHANNEL" = "PREVIEW" を設定します。プレビューの詳細については、パイプラインのプロパティ リファレンスを参照してください。

以下の例では、LakeflowパイプラインでAuto Loaderを使用してOneDriveファイルを読み取る方法を示します。

Python

Python

from pyspark import pipelines as dp

# Incrementally ingest new PDF files
@dp.table
def onedrive_pdf_table():
  return (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "binaryFile")
    .option("databricks.connection", "my_onedrive_conn")
    .option("pathGlobFilter", "*.pdf")
    .load("https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents")
  )

# Incrementally ingest CSV files with automatic schema inference and evolution
@dp.table
def onedrive_csv_table():
  return (spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .option("databricks.connection", "my_onedrive_conn")
      .option("pathGlobFilter", "*.csv")
      .option("cloudFiles.inferColumnTypes", True)
      .option("header", True)
      .load("https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/IoT_Logs")
  )

SQL

SQL

-- Incrementally ingest new PDF files
CREATE OR REFRESH STREAMING TABLE onedrive_pdf_table
AS SELECT * FROM STREAM read_files(
  "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents",
  format => "binaryFile",
  `databricks.connection` => "my_onedrive_conn",
  pathGlobFilter => "*.pdf");

-- Incrementally ingest CSV files with automatic schema inference and evolution
CREATE OR REFRESH STREAMING TABLE onedrive_csv_table
AS SELECT * FROM STREAM read_files(
  "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/IoT_Logs",
  format => "csv",
  `databricks.connection` => "my_onedrive_conn",
  pathGlobFilter => "*.csv",
  header => "true");

増分取り込み（使用： COPY INTO

COPY INTO は、Delta テーブルへのファイルのべき等な増分読み込みを提供します。COPY INTO の使用方法の詳細については、COPY INTO を使用した一般的なデータ読み込みパターンを参照してください。

SQL

CREATE TABLE IF NOT EXISTS onedrive_pdf_table;
CREATE TABLE IF NOT EXISTS onedrive_csv_table;

-- Incrementally ingest new PDF files
COPY INTO onedrive_pdf_table
  FROM "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents"
  FILEFORMAT = BINARYFILE
  PATTERN = '*.pdf'
  FORMAT_OPTIONS ('databricks.connection' = 'my_onedrive_conn')
  COPY_OPTIONS ('mergeSchema' = 'true');

-- Incrementally ingest CSV files with automatic schema inference and evolution
COPY INTO onedrive_csv_table
  FROM "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents/IoT_Logs"
  FILEFORMAT = CSV
  PATTERN = '*.csv'
  FORMAT_OPTIONS ('databricks.connection' = 'my_onedrive_conn', 'header' = 'true', 'inferSchema' = 'true')
  COPY_OPTIONS ('mergeSchema' = 'true');

変更追跡による効率的な増分取り込み

defaultでは、Auto Loaderは更新のたびにドライブをリスト化することで、OneDrive内の新しいファイルを検出します。大規模なドライブの場合は、代わりに Microsoft Graph チェンジフィードを使用して変更を追跡できます。これにより、ドライブ全体を再リスト化するのではなく、前回の更新以降に追加または変更されたファイルのみが返されます。これを有効にするには、cloudFiles.readChangeFeed オプションを true に設定します。

Python

df = (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "binaryFile")
    .option("cloudFiles.readChangeFeed", "true")
    .option("databricks.connection", "my_onedrive_conn")
    .option("cloudFiles.schemaLocation", <path to a schema location>)
    .load("https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents")
)

cloudFiles.readChangeFeedを使用する場合は、次の点に注意してください。

• binaryFile形式が必要です。他の形式では変更追跡を利用できません。
• ロードパスは、ドライブ、またはドライブ内のディレクトリに解決される必要があります。単なる個人サイトやテナントルートはサポートされていません。
• 明示的なスキーマを提供することはできません。出力スキーマは、ファイル形式によって自動的に決定されます。

非構造化ファイルを解析

binaryFile形式の標準OneDriveコネクタを使用してOneDriveから非構造化ファイル（PDF、Wordドキュメント、PowerPointファイルなど）を取り込む場合、ファイルコンテンツは生のバイナリデータとして保存されます。RAG、検索、分類、ドキュメント理解などのAIワークロード用にこれらのファイルを準備するには、ai_parse_documentを使用してバイナリコンテンツを構造化されたクエリ可能な出力に解析できます。

次の例は、documents という名前のブロンズ Delta テーブルに格納されている非構造化ドキュメントを解析し、解析されたコンテンツを含む新しい列を追加する方法を示しています。

SQL

CREATE TABLE documents AS
SELECT * FROM read_files(
  "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents",
  `databricks.connection` => "my_onedrive_conn",
  format => "binaryFile",
  pathGlobFilter => "*.{pdf,docx}",
  schemaEvolutionMode => "none"
);
SELECT *, ai_parse_document(content, map('version', '2.0')) AS parsed_content
FROM documents;

Lakeflow パイプライン内で ai_parse_document を使用して、増分解析を有効にすることもできます。例えば、未加工のバイナリファイルを取り込むブロンズストリーミングテーブルを定義し、次に解析されたコンテンツを新しい列に格納する2つ目のストリーミングテーブルを定義できます：

SQL

-- Bronze: incrementally ingest the raw documents as binary
CREATE OR REFRESH STREAMING TABLE onedrive_documents
AS SELECT * FROM STREAM read_files(
  "https://mytenant-my.sharepoint.com/personal/user_mytenant_onmicrosoft_com/Documents",
  `databricks.connection` => "my_onedrive_conn",
  format => "binaryFile",
  pathGlobFilter => "*.{pdf,docx}");

-- Silver: land the parsed content in a new column
CREATE OR REFRESH STREAMING TABLE onedrive_documents_parsed
AS SELECT
  *,
  ai_parse_document(content, map('version', '2.0')) AS parsed_content
FROM STREAM onedrive_documents;

parsed_content列には、ダウンストリームのAIパイプラインで直接使用できる、抽出されたテキスト、テーブル、Layout情報、およびメタデータが含まれています。

両方のテーブルがストリーミングテーブルであるため、パイプラインが更新されるたびに新しく取り込まれたOneDriveファイルが自動的に解析され、解析された出力は受信データと同期された状態に保たれます。

詳細については、サポートされている形式と高度なオプションについて ai_parse_document を参照してください。これには、出力スキーマをピン留めする version オプションが含まれます。

制限事項

標準のOneDriveコネクタには、以下の制限があります。

• このコネクタは、単一ユーザーの個人用 OneDrive ( マイファイル ) から取り込みます。SharePoint サイト、チームサイト、および共有ドキュメントライブラリはサポートされていません。それらには SharePoint コネクタ を使用してください。
• ベアホストURLおよびドライブ以外のパスはサポートされていません。
• OneDriveのリストおよびサイトページはサポートされていません。このコネクタは、ユーザーのOneDriveドライブに保存されているファイルのみを取り込みます。
• pathGlobFilterオプションを使用して、名前でファイルをフィルタリングできます。フォルダーパスベースのフィルターはサポートされていません。
• cloudFiles.readChangeFeedを使用した変更追跡には、binaryFile形式と、ドライブまたはドライブ内のディレクトリに解決されるロードパスが必要です。変更追跡による効率的な増分取り込みを参照してください。
• OneDriveへの書き戻しはサポートされていません。このコネクタは読み取り専用です。
• Auto Loader cleanSource（取り込み後にソースのファイルを削除またはアーカイブする）はサポートされていません。

その他のリソース

• 高度なストリーミング取り込みパターンのための Auto Loader について学ぶ
• べき等な増分ロードのための COPY INTO を確認する
• 非構造化ファイルの解析については、ai_parse_documentを確認してください
• SharePointサイトから取り込むためのSharePointコネクタと比較してください
• 取り込みワークフローを自動化するためにジョブスケジューリングを設定します