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

備考

ベータ版

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

注記

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

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

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

OneDriveからのデータ取り込み方法を選択します

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

オプション	説明
管理対象のOneDriveコネクター	エンタープライズアプリケーション向けのフルマネージドコネクタで、データをDeltaテーブルに取り込み、ソースとの同期を維持します。
SQLおよびSpark API	read_files、spark.read、COPY INTO、Auto Loaderなどのバッチおよびストリーミング APIsを使用して、SQL、PySpark、またはLakeFlow Pipelinesでカスタム取り込みパイプラインを構築できます。取り込み中に複雑な変換を実行できる柔軟性を提供する一方、パイプラインの管理と保守における責任が増します。

オプション	説明
管理対象のOneDriveコネクター	エンタープライズアプリケーション向けのフルマネージドコネクタで、データをDeltaテーブルに取り込み、ソースとの同期を維持します。
SQLおよびSpark API	read_files、spark.read、COPY INTO、Auto Loaderなどのバッチおよびストリーミング APIsを使用して、SQL、PySpark、またはLakeFlow Pipelinesでカスタム取り込みパイプラインを構築できます。取り込み中に複雑な変換を実行できる柔軟性を提供する一方、パイプラインの管理と保守における責任が増します。

主な機能

SQL および Spark API で提供される機能：

• ユーザーの個人用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）認証を使用します。SQL API と Spark API のいずれを使用する場合でも、接続のセットアッププロセスは、マネージド OneDrive コネクタを使用する場合と同じです。

OAuth U2M：Databricks管理およびOAuth U2M：カスタム管理のオプションを含む、接続設定の完全な手順については、OneDrive接続の作成を参照してください。

接続はサインインしているユーザーとしてファイルを読み取るため、ユーザーが自分の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は、SharePointのホスト（mytenant.sharepoint.com）とは異なる個人の -my ホスト（例：mytenant-my.sharepoint.com）を使用します。SharePointサイトへの共有Linkは受け付けられません。SharePointから取り込むには、代わりにSharePointからのファイルの取り込みを参照してください。

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

例

SQL および Spark API を使用して 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

注記

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

非構造化ファイルを解析

SQL および Spark API と binaryFile 形式を使用して 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;

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

LakeFlow Pipelines を使用した増分解析

LakeFlow Pipelines内動詞で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;

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

サポートされている形式と詳細オプション（出力スキーマをピン留めする version オプションを含む）については、ai_parse_document 関数を参照してください。

制限事項

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

• OneDrive の取り込みでは、単一ユーザーの個人の OneDrive ( My files ) から読み取りを行います。SharePoint サイト、チームサイト、および共有ドキュメント ライブラリはサポートされていません。それらについては、SharePoint からのファイルの取り込みを参照してください。
• OneDrive のリストおよびサイトページはサポートされていません。ユーザーの OneDrive ドライブに保存されているファイルのみが取り込まれます。
• pathGlobFilterオプションを使用して、名前でファイルをフィルタリングできます。フォルダーパスベースのフィルターはサポートされていません。
• OneDrive への書き戻しはサポートされていません。取り込みは読み取り専用です。
• Auto Loader cleanSource（取り込み後にソースのファイルを削除またはアーカイブする）はサポートされていません。

その他のリソース

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