Google Drive ファイルを Databricks に取り込む

備考

ベータ版

この機能はベータ版です。ワークスペース管理者は、 プレビュー ページからこの機能へのアクセスを制御できます。「Databricks プレビューの管理」を参照してください。

LakeFlow Connectの標準の Google ドライブ コネクタを使用すると、 Databricks SparkおよびSQL関数 ( read_files 、 spark.read 、 COPY INTO 、およびAuto Loader ) を使用して、Google ドライブ内のファイルからSparkデータフレーム、マテリアライズドビュー、ストリーミング テーブルを直接作成できます。

このアプローチを使用すると、一般的なファイル取り込みユースケース向けのカスタム パイプラインを作成できます。

• ストリーミング ファイルの取り込み (非構造化): 多数のソース ファイル (PDF、Google ドキュメント、Google スライドなど) をバイナリ データとして単一のターゲット テーブルに取り込みます。RAG パイプラインに最適です。
• ストリーミング ファイルの取り込み (構造化): 多数のソース ファイル (CSV ファイルや JSON ファイルなど) を 1 つの構造化されたターゲット テーブルにマージします。
• バッチ ファイルの取り込み: 単一の特定のファイル (Google シートなど) またはファイルのバッチを 1 つのターゲット テーブルに取り込みます。

次のインターフェースがサポートされています:

• 宣言型自動化バンドル
• Databricks API
• Databricks SDK
• Databricks CLI

制限事項

• コネクタは API のみであり、Databricks UI でのパイプラインの作成はサポートされていません。
• pathGlobFilterオプションを使用すると、名前 (例: pathGlobFilter => '*.csv' ) でファイルをフィルターできます。組み込みの Google 形式（Google ドキュメントや Google スプレッドシートなど）は、このオプションを使用してフィルタリングできません。フォルダー パス フィルタリングもサポートされていません。
• サポートされていない形式には、Google フォーム、Google サイト、Google Jams、Google Vids などがあります。取り込みプロセスでは、サポートされていない形式はスキップされます。
• 設定recursiveFileLookup=falseは Spark バッチ読み取りではサポートされていません。recursiveFileLookup=falseを使用するとrecursiveFileLookup=trueと同じように動作します。

要件

始める前に、次のことを確認してください。

• Unity カタログ対応のワークスペース。
• CREATE CONNECTION 権限。
• Databricks Runtime 17.3 以上。
• Google スプレッドシートまたは Excel ファイルを取り込む場合は、Excel ファイル形式のベータ機能が有効になります。Excel ファイルの読み取りを参照してください。
• Google クラウド プロジェクトを作成するために必要な権限を持つ Google アカウント。

OAuth 2.0を構成する

Google クラウド プロジェクトをセットアップし、Google Drive APIを有効にする

1. Google クラウド コンソールに移動します。
2. 新しいプロジェクトを作成します。2 要素認証を設定するように求められる場合があります。
3. APIsとサービス] > [ライブラリ] に移動します。
4. 「Google Drive API」を検索します。
5. Google Drive API を 選択します。
6. [有効にする] を選択します。

プロジェクトのOAuth同意画面を設定する

1. Google Cloud コンソールのホーム画面で、 APIsとサービス] > OAuth同意画面] に移動します。 「Google Auth Platform はまだ構成されていません」というメッセージが表示されます。
2. [開始する] を選択します。
3. アプリ情報 セクションに入力します。アプリ名に任意の名前を入力します (例: Databricks connection )。サポート電子メールには、組織内の任意の電子メールを使用できます。
4. 「次へ」 を選択します。
5. [Audience] セクションで [External] を選択し、 [Next] を選択します。
6. 連絡先情報 セクションに入力し、 [次へ] を選択します。
7. Google API サービス ユーザー データ ポリシーを確認し、 [作成] を選択します。
8. Google Auth Platform > データ アクセス に戻ります。
9. スコープの追加または削除を 選択します。
10. [手動でスコープを追加] セクションで次のスコープを追加し、 [テーブルに追加] を選択して、 [更新] を選択します。 https://www.googleapis.com/auth/drive.readonly
11. [保存] を選択します。

OAuth 2.0クライアント資格情報を作成する

1. Google Cloud コンソールのホーム画面で、 APIsとサービス] > [認証情報] に移動します。
2. [資格情報の作成] > [OAuth クライアント ID] を選択します。
3. Web アプリケーション を選択し、カスタム名を設定します。
4. [承認済みリダイレクト URI] で、 [URI の追加] を選択します。
5. リダイレクト URI を<databricks-instance-url>/login/oauth/google.htmlに追加し、 <databricks-instance-url> Databricks インスタンス URL に置き換えます。例えば： https://instance-name.databricks.com/login/oauth/google.html
6. [作成] を選択します。資格情報を含むダイアログが表示されます。
7. 次の値を記録します。あるいは、次の情報が含まれる OAuth クライアント JSON ファイルをダウンロードすることもできます。
   • クライアント ID (形式: 0123******-********************************.apps.googleusercontent.com )
   • クライアントシークレット (形式: ABCD**-**************************** )

プロジェクトにテストユーザーを追加する

1. Google Auth Platform > Audience に移動します。
2. [テスト ユーザー] の下で、 [ユーザーの追加] を選択します。
3. 接続の作成に使用する Google アカウントの電子メール アドレスを追加します。

接続を作成する

1. Databricksワークスペースで、 [カタログ] > [外部ロケーション] > [接続] > [接続の作成] を選択します。

2. 接続の設定 ウィザードの 接続の基本 ページで、一意の 接続名 を指定します。

3. [接続タイプ] ドロップダウン メニューで、 Google Drive を検索して選択します。

4. （オプション）コメントを追加します。

5. 「次へ」 を選択します。

6. 認証 ページで、次のように入力します。

   • OAuthスコープ : https://www.googleapis.com/auth/drive.readonly
   • クライアント シークレット : OAuth 2.0 クライアント資格情報の作成からのクライアント シークレット。
   • クライアント ID : OAuth 2.0 クライアント資格情報の作成からのクライアント ID。

7. [Google でサインイン] を 選択し、 [プロジェクトにテスト ユーザーを追加する] から Google アカウントでサインインします。

8. [続行] を選択し、もう一度 [続行] を選択します。

9. Databricks ワークスペースにリダイレクトされたら、 [接続の作成] を選択します。

Google ドライブからファイルを取り込む

コネクタとread_files ( Databricks SQL )、 Auto Loader ( .readStreamとcloudFiles )、 COPY INTO 、 spark.readを使用して、Google ドライブからファイルを取り込みます。 次の値を指定する必要があります。

• パスとしての Google ドライブの URL。
• databricks.connectionデータソース オプションを使用したUnity Catalog接続。
• アクセスする Google ドライブ リソースの URL。URL は特定のファイル、フォルダー、またはドライブ全体を参照できます。例えば：
  • https://docs.google.com/spreadsheets/d/12345/edit?random_query_params_here
  • https://drive.google.com/drive/u/0/folders/12345
  • https://docs.google.com/document/d/12345/edit
  • https://drive.google.com/file/d/1kiXnHmU4Y8X66ijULky5EPDNCGtT14Ps/view?usp=drive_link
  • https://drive.google.com/drive/
  • https://drive.google.com/drive/my-drive
  • https://drive.google.com/

Auto Loaderを使用した Google ドライブ ファイルのストリーム

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

Python

# Incrementally ingest new PDF files
df = (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "binaryFile")
    .option("databricks.connection", "my_gdrive_conn")
    .option("cloudFiles.schemaLocation", <path to a schema location>)
    .option("pathGlobFilter", "*.pdf")
    .load("https://drive.google.com/drive/folders/1a2b3c4d...")
    .select("*", "_metadata")
)

# Incrementally ingest CSV files with automatic schema inference and evolution
df = (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("databricks.connection", "my_gdrive_conn")
    .option("pathGlobFilter", "*.csv")
    .option("inferColumnTypes", True)
    .option("header", True)
    .load("https://drive.google.com/drive/folders/1a2b3c4d...")
)

Spark バッチ読み取りを使用して Google ドライブのファイルを読み取る

次の例は、 spark.read関数を使用して Python で Google ドライブ ファイルを取り込む方法を示しています。サポートされているファイル形式と Spark リーダー オプションの一覧については、Apache Spark ドキュメントの「汎用ファイル ソース オプション」を参照してください。

設定recursiveFileLookup=falseは Spark バッチ読み取りではサポートされていません。recursiveFileLookup=falseを使用するとrecursiveFileLookup=trueと同じように動作します。

Python

# Read unstructured data as binary files
df = (spark.read
        .format("binaryFile")
        .option("databricks.connection", "my_gdrive_conn")
        .option("recursiveFileLookup", True)
        .option("pathGlobFilter", "*.pdf")
        .load("https://drive.google.com/drive/folders/1a2b3c4d..."))

# 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_gdrive_conn")
        .option("pathGlobFilter", "*.csv")
        .option("recursiveFileLookup", True)
        .option("inferSchema", True)
        .option("header", True)
        .load("https://drive.google.com/drive/folders/1a2b3c4d..."))

構造化された単一の Google ドライブ ファイルを読み取る

次の例では、Google スプレッドシートの 1 つのタブを読み取り、それを DataFrame に読み込みます。一般的な解析オプションをいくつか示します。

Excel ファイルと Google スプレッドシートでサポートされている解析オプションの完全なリストについては、 「Excel ファイルの読み取り」をご覧ください。サポートされているその他のすべてのファイル形式と Spark リーダー オプションの完全なリストについては、Apache Spark ドキュメントの「汎用ファイル ソース オプション」を参照してください。

Python

df = (spark.read
  .format("excel")  # use 'excel' for Google Sheets
  .option("databricks.connection", "my_gdrive_conn")
  .option("headerRows", 1) # optional
  .option("inferColumns", True) # optional
  .option("dataAddress", "Sheet1!A1:Z10")  # optional
  .load("https://docs.google.com/spreadsheets/d/9k8j7i6f..."))

Spark SQL を使用して Google ドライブのファイルを読み取る

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

SQL

-- Read pdf files
CREATE TABLE my_table AS
SELECT * FROM read_files(
  "https://drive.google.com/drive/folders/1a2b3c4d...",
  `databricks.connection` => "my_gdrive_conn",
  format => "binaryFile",
  pathGlobFilter => "*.pdf", -- optional. Example: only ingest PDFs
  schemaEvolutionMode => "none"
);

-- Read a Google Sheet and range
CREATE TABLE my_sheet_table AS
SELECT * FROM read_files(
  "https://docs.google.com/spreadsheets/d/9k8j7i6f...",
  `databricks.connection` => "my_gdrive_conn",
  format => "excel",
  headerRows => 1,  -- optional
  dataAddress => "Sheet1!A2:D10", -- optional
  schemaEvolutionMode => "none"
);

LakeFlow Spark宣言型パイプラインを使用して Google ドライブ ファイルを取り込む

注記

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

この例では、 LakeFlow Spark宣言型パイプラインでAuto Loaderを使用して Google ドライブ ファイルを読み取る方法を示します。 詳細については、 LakeFlow Spark宣言型パイプラインを参照してください。

Databricks SQL

SQL

-- Incrementally ingest new PDF files
CREATE OR REFRESH STREAMING TABLE gdrive_pdf_table
AS SELECT * FROM STREAM read_files(
  "https://drive.google.com/drive/folders/1a2b3c4d...",
  format => "binaryFile",
  `databricks.connection` => "my_gdrive_conn",
  pathGlobFilter => "*.pdf");

-- Incrementally ingest CSV files with automatic schema inference and evolution
CREATE OR REFRESH STREAMING TABLE gdrive_csv_table
AS SELECT * FROM STREAM read_files(
  "https://drive.google.com/drive/folders/1a2b3c4d...",
  format => "csv",
  `databricks.connection` => "my_gdrive_conn",
  pathGlobFilter => "*.csv",
  "header", "true");

-- Read a specific Excel file from Google Drive in a materialized view
CREATE OR REFRESH MATERIALIZED VIEW gdrive_excel_table
AS SELECT * FROM read_files(
  "https://docs.google.com/spreadsheets/d/9k8j7i6f...",
  `databricks.connection` => "my_gdrive_conn",
  format => "excel",
  headerRows => 1,  -- optional
  dataAddress => "Sheet1!A2:D10", -- optional
  `cloudFiles.schemaEvolutionMode` => "none"
);

Python

Python

from pyspark import pipelines as dp

# Incrementally ingest new PDF files
@dp.table
def gdrive_pdf_table():
  return (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "binaryFile")
    .option("databricks.connection", "my_gdrive_conn")
    .option("pathGlobFilter", "*.pdf")
    .load("https://drive.google.com/drive/folders/1a2b3c4d...")
  )

# Incrementally ingest CSV files with automatic schema inference and evolution
@dp.table
def gdrive_csv_table():
  return (spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .option("databricks.connection", "my_gdrive_conn")
      .option("pathGlobFilter", "*.csv")
      .option("inferColumnTypes", True)
      .option("header", True)
      .load("https://drive.google.com/drive/folders/1a2b3c4d...")
  )

# Read a specific Excel file from Google Drive in a materialized view
@dp.table
def gdrive_excel_table():
  return (spark.read.format("excel")
    .option("databricks.connection", "my_gdrive_conn")
    .option("headerRows", 1) # optional
    .option("inferColumnTypes", True) # optional
    .option("dataAddress", "Sheet1!A1:M20") # optional
    .load("https://docs.google.com/spreadsheets/d/9k8j7i6f..."))

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

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

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

SQL

CREATE TABLE documents AS
SELECT *, _metadata FROM read_files(
  "https://drive.google.com/drive/folders/1a2b3c4d...",
  `databricks.connection` => "my_gdrive_conn",
  format => "binaryFile",
  pathGlobFilter => "*.{pdf,jpeg}"
);
SELECT *, ai_parse_document(content) AS parsed_content
FROM documents;

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

LakeFlow Spark宣言型パイプラインを使用した増分解析

LakeFlow Spark宣言型パイプライン内でai_parse_document使用して、増分解析を有効にすることもできます。 Google ドライブから新しいファイルがストリームされると、パイプラインの更新時に自動的に解析されます。

たとえば、新しく取り込まれたドキュメントを継続的に解析するマテリアライズドビューを定義できます。

SQL

CREATE OR REFRESH STREAMING TABLE documents
AS SELECT *, "_metadata" FROM STREAM read_files(
  "https://drive.google.com/drive/folders/1a2b3c4d...",
  format => "binaryFile",
  `databricks.connection` => "my_gdrive_conn",
  pathGlobFilter => "*.{pdf,jpeg}");

CREATE OR REFRESH MATERIALIZED VIEW documents_parsed
AS
SELECT *, ai_parse_document(content) AS parsed_content
FROM documents;

このアプローチにより、次のことが保証されます。

• 新しく取り込まれたGoogleドライブファイルは、マテリアライズドビューが更新されるたびに自動的に解析されます。
• 解析された出力は入力データと同期されます
• 下流のAIパイプラインは常に最新のドキュメント表現で動作します

サポートされている形式と詳細オプションについては、 ai_parse_document関数を参照してください。

組み込みのGoogle形式の処理方法

組み込みの Google ファイル (ドキュメント、スプレッドシート) を手動でエクスポートする必要はありません。コネクタは、取り込み中にそれらを自動的にオープン形式にエクスポートします。

Googleフォーマット	エクスポート形式（デフォルト）
Googleドキュメント	application/vnd.openxmlformats-officedocument.wordprocessingml.document （DOCX）
Googleスプレッドシート	application/vnd.openxmlformats-officedocument.spreadsheetml.sheet （XLSX）
Googleスライド	application/vnd.openxmlformats-officedocument.presentationml.presentation （PPTX）
Google 描画	application/pdf （PDF）

Googleドライブのエクスポート形式の設定

spark.conf.set()を使用して Spark 構成を設定することで、Google ドライブのネイティブ ファイルのエクスポート方法を設定できます。これらの設定により、Google ドキュメント、スプレッドシート、スライド、図形描画をエクスポートするときに使用するMIMEタイプが決まります。

構成キー:

• fs.gdrive.format.document.export: Google ドキュメントのエクスポート形式。
• fs.gdrive.format.spreadsheet.export: Google スプレッドシートのエクスポート形式。
• fs.gdrive.format.presentation.export: Google スライドのエクスポート形式。
• fs.gdrive.format.drawing.export: Google 図形描画のエクスポート形式。

サポートされているエクスポート形式の完全なリストについては、Google ワークスペースのドキュメントの「Google ワークスペース ドキュメントの MIME タイプのエクスポート」を参照してください。

次の例では、Google ドキュメント ファイルを TXT としてエクスポートします。

Python

spark.conf.set("fs.gdrive.format.document.export", "text/plain")
df = spark.read.text("https://docs.google.com/document/d/1a2b3c4d...")

次の例では、Google スプレッドシート ファイルを CSV としてエクスポートします。

Python

spark.conf.set("fs.gdrive.format.spreadsheet.export", "text/csv")
df = spark.read.option("header", "true").csv("https://docs.google.com/spreadsheets/d/1a2b3c4d...")

スキーマ

バイナリファイル形式

format => 'binaryFile'を使用すると、結果のテーブルのスキーマは次のようになります。

• path (文字列): ファイルの完全な URL。
• modificationTime (タイムスタンプ): ファイルの最終更新時刻。
• length (long): バイト単位のファイルサイズ。
• content (バイナリ): ファイルの生のバイナリ コンテンツ。

_metadata列

file_name 、 file_path 、 file_size 、 file_modification_timeなどの_metadata列を持つ入力ファイルのメタデータ情報を取得できます。_metadata列は非表示の列であり、すべての入力ファイル形式で使用できます。返される DataFrame に_metadata列を含めるには、ソースを指定する読み取りクエリで明示的に選択する必要があります。詳細については、 「ファイルのメタデータ列」を参照してください。

Google ドライブのネイティブ ファイル（Google ドキュメント、Google スプレッドシート、Google スライドなど）の場合、 file_sizeフィールドは、エクスポートされた形式（DOCX、XLSX、PPTX など）のファイルのサイズではなく、Google ドライブに保存されているファイルのサイズを示します。

_metadataを選択する例:

SQL

SELECT *, _metadata FROM read_files(
  "https://drive.google.com/",
  `databricks.connection` => "my_connection",
  format => "binaryFile"
);

よくある質問

複数の構造化ファイル（複数の Google スプレッドシートなど）のフォルダーがあります。各シートまたはファイルを独自の Delta テーブルとしてロードするにはどうすればよいですか?

各ファイルを独自の Delta テーブルに取り込むには、新しい取り込みクエリを作成する必要があります。

ファイルにはカスタム解析が必要です。ファイルが正しく読み取られることを確認するために、これらの解析を提供するにはどうすればよいですか?

コネクタは、Auto Loader、 COPY INTO 、および Spark で使用可能なすべてのファイル形式オプションをサポートしています。詳細については、以下を参照してください。

• Auto Loaderオプション
• 基本オプション
• データソース - Spark

サブフォルダー内のファイルは再帰的に取り込まれますか?

Auto Loader APIs ( spark.readStreamおよびread_files ) を使用すると、すべてのサブフォルダーが再帰的に検出され、取り込まれます。 これはバッチspark.readにも当てはまり、指定されたフォルダー パス内のファイルは常に再帰的に読み取られます。

私の Google スプレッドシートには多くの不規則性があり、特定の解析とセル範囲の抽出が必要です (たとえば、シートごとに複数のテーブルがある)。スキーマを自動的に推論することはできません。これをどう処理すればいいでしょうか?

Excel ファイル形式の解析オプションを使用して、Google スプレッドシート ファイルを希望の形式に解析できます。Excel ファイルの読み取りを参照してください。

あるいは、Auto Loader、PySpark、またはread_filesでスキーマ推論を無効にすることもできます。結果のテーブルにはデフォルトの列名が付けられ、すべてのデータ型は文字列にキャストされ、テーブルはスパースになる可能性があります。その後、下流で必要な変換を実行できます。

次のステップ

• 一般的なデータ読み込みパターンを調べる
• ai_parse_document関数をレビューする