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

Lakeflow Connectでは、Azure Databricks SparkやSQL関数(read_files、spark.read、COPY INTO、Autoloader)を使って、Googleドライブのファイルから直接Sparkデータフレーム、マテライズドビュー、ストリーミングテーブルを作成できます。

この方法では、一般的なファイル インジェストのユース ケース用のカスタム パイプラインを作成できます。

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

これらのインターフェイスがサポートされています。

• 宣言型オートメーション バンドル
• Databricks API
• Databricks SDK
• Databricks コマンドラインインターフェース (CLI)

Googleドライブからの取り込み方法を選択

Lakeflow ConnectはGoogleドライブからのインジェスト方法を2つ提供しています。 どちらも Google Drive のデータにアクセスしますが、管理レベルが異なります。

制限事項

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

Requirements

開始する前に、次のことを確認します。

• Unity カタログ対応ワークスペース。
• CREATE CONNECTION Google ドライブ接続を作成するための特権、または クラスターアクセスモードに基づいて既存のものを使用するための適切な特権:
  • 専用アクセス モード: MANAGE CONNECTION。
  • 標準アクセス モード: USE CONNECTION。
• Databricks Runtime 17.3 以降。
• Google Cloud プロジェクトを作成するために必要なアクセス許可を持つ Google アカウント。

OAuth 2.0 の構成

Google Cloud プロジェクトを設定し、Google Drive API を有効にする

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

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

1. Google Cloud コンソールのホーム画面で、 API とサービス > OAuth 同意画面に移動します。 "Google 認証プラットフォームがまだ構成されていません" というメッセージが表示されます。
2. [開始] を選択します。
3. [ アプリ情報 ] セクションに入力します。 アプリ名の任意の名前を入力します (例: Databricks connection)。 サポート メールには、組織内の任意の電子メールを指定できます。
4. 次へを選択します。
5. オーディエンスセクションでオーディエンスを選択し、次に選択します:
   • 内部:Google Workspace組織内のユーザーのみがアプリを使い、Googleのポリシーで社内アプリが許可されている場合、このオプションを選択してください。
   • 外部:Google Workspace組織外の人がアプリを使用している場合は、このオプションを選択してください。
6. [連絡先情報] セクションに入力し、[次へ] を選択します。
7. Google API Services ユーザー データ ポリシーを確認し、[ 作成] を選択します。
8. Google 認証プラットフォーム > データ アクセスに戻ります。
9. [ スコープの追加または削除] を選択します。
10. [手動でスコープを追加する] セクションで次 のスコープを追加 し、[ テーブルに追加] を選択し、[ 更新] を選択します。 https://www.googleapis.com/auth/drive.readonly
11. 保存を選びます。

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

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

外部アプリのテストユーザーを追加する

外部を選択している場合は、接続作成に使ったGoogleアカウントをテストユーザーとして追加してください。 Internal を選択した場合は、このステップをスキップしてください。

1. Google 認証プラットフォーム>対象ユーザーに移動します。
2. [ ユーザーのテスト] で、[ ユーザーの追加] を選択します。
3. 接続の作成に使用する Google アカウントのメール アドレスを追加します。

接続を作成する

1. Azure Databricks ワークスペースで、Catalog > 外部の場所 > Connections > 接続の作成 を選択します。

2. 接続のセットアップ ウィザードの [接続の基本] ページで、一意の接続名を指定します。

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

4. (省略可能)コメントを追加します。

5. 次へを選択します。

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

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

7. 「Googleでサインイン」を選択し、アプリを使えるGoogleアカウントでサインインしてください。

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

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

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

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

• パスとしての Google ドライブの URL。
• databricks.connection データ ソース オプションを使用した Unity カタログ接続。
• アクセスする 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/

自動ローダーを使用して Google ドライブ ファイルをストリーミングする

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

# 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-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("cloudFiles.schemaLocation", "<path-to-schema-location>")
    .option("pathGlobFilter", "*.csv")
    .option("cloudFiles.inferColumnTypes", True)
    .option("header", True)
    .load("https://drive.google.com/drive/folders/1a2b3c4d...")
)

Spark バッチ読み取りを使用して Google Drive ファイルを読み取る

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

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

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

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

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

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

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

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

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

-- 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 パイプラインを使用して Google Drive ファイルを取り込む

以下の例は、LakeflowパイプラインのAuto Loaderを使ってGoogle Driveファイルを読み取る方法を示しています。 LakeflowのSpark Declarative Pipelinesを参照してください。

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

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("cloudFiles.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("dataAddress", "Sheet1!A1:M20") # optional
    .load("https://docs.google.com/spreadsheets/d/9k8j7i6f..."))

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

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

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

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, map('version', '2.0')) AS parsed_content
FROM documents;

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

Lakeflow パイプラインを使用した増分解析

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

例えば、生のバイナリファイルを取り込むブロンズストリーミングテーブルを定義し、解析したコンテンツを新しい列に格納する2つ目のストリーミングテーブルを定義できます:

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 STREAMING TABLE documents_parsed
AS SELECT
  *,
  ai_parse_document(content, map('version', '2.0')) AS parsed_content
FROM STREAM documents;

この方法により、次のことが保証されます。

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

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

組み込み Google フォーマットの取り扱い方法

組み込みの Google ファイル (Docs、スプレッドシート) を手動でエクスポートする必要はありません。 インジェスト時に自動的にオープンフォーマットにエクスポートされます。

Google ドライブのエクスポート形式の構成

spark.conf.set()を使用して Spark 構成を設定することで、Google Drive ネイティブ ファイルのエクスポート方法を構成できます。 これらの構成により、Google Docs、スプレッドシート、スライド、および図面をエクスポートするときに使用する MIME の種類が決まります。

構成キー:

• fs.gdrive.format.document.export: Google Docs のエクスポート形式。
• fs.gdrive.format.spreadsheet.export: Google スプレッドシートのエクスポート形式。
• fs.gdrive.format.presentation.export: Google スライドのエクスポート形式。
• fs.gdrive.format.drawing.export: Google 図面のエクスポート形式。

サポートされているエクスポート形式の完全な一覧については、 Google Workspace ドキュメントの「Google Workspace ドキュメントの MIME の種類をエクスポート する」を参照してください。

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

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

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

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

Schemas

バイナリファイルフォーマット

format => 'binaryFile'を使用する場合、結果のテーブルには次のスキーマがあります。

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

_metadata列

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

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

_metadataを選択する例:

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

Google Drive メタデータ列

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

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

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

スキーマ

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

例

次の例では、 _gdrive_metadata 列を読み取りクエリに含める方法、列から特定のフィールドを選択する方法、および VARIANT フィールドから値を抽出する方法を示します。

Python

df = (spark.read
  .format("binaryFile")
  .option("databricks.connection", "my_gdrive_conn")
  .load("https://drive.google.com/drive/folders/1a2b3c4d...")
  .select("*", "_metadata", "_gdrive_metadata"))

SQL

SELECT *, _gdrive_metadata
FROM read_files(
  "https://drive.google.com/drive/folders/1a2b3c4d...",
  `databricks.connection` => "my_gdrive_conn",
  format => "binaryFile"
);

_gdrive_metadata構造体から特定のフィールドを選択します。

df = (spark.read
  .format("binaryFile")
  .option("databricks.connection", "my_gdrive_conn")
  .load("https://drive.google.com/drive/folders/1a2b3c4d...")
  .select("_gdrive_metadata.id", "_gdrive_metadata.md5_checksum"))

properties キャスト演算子を使用して、additional_metadataまたはVARIANT::フィールドから値を抽出します。

SELECT
  *,
  _gdrive_metadata.properties:department::STRING AS department
FROM read_files(
  "https://drive.google.com/drive/folders/1a2b3c4d...",
  `databricks.connection` => "my_gdrive_conn",
  format => "binaryFile"
);

FAQ

複数の構造化ファイル(たとえば、複数のGoogleスプレッドシート)のフォルダがあります。 各シートまたはファイルを独自の Delta テーブルとして読み込む方法

ファイルごとに新しいインジェスト クエリを作成して、独自の Delta テーブルに取り込む必要があります。

ファイルにはカスタム解析が必要です。 これらの解析パラメーターを指定して、ファイルが正しく読み取られるようにするにはどうすればよいですか?

SQLおよびSpark APIは、Auto Loader、 COPY INTO、Sparkで利用可能なすべてのファイルフォーマットオプションをサポートしています。 詳細については、次を参照してください。

• 自動ローダー
• 基本オプション
• データ ソース - Spark

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

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

Google シートには多くの凹凸があり、特定の解析とセル範囲抽出 (たとえば、シートごとに複数のテーブル) が必要です。 スキーマを自動的に推論することはできません。 これを処理する方法

Excelファイル形式の解析オプションを使用して、Google シート ファイルを目的の形式に解析できます。 「Excel ファイルの読み取りとストリーム配信」を参照してください。

または、自動ローダー、PySpark、または read_filesでスキーマ推論を無効にすることもできます。 結果のテーブルには既定の列名があり、すべてのデータ型が文字列にキャストされ、テーブルがスパースである可能性があります。 その後、ダウンストリームで必要な変換を実行できます。

その他のリソース

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