概要

本記事では、Google Colab で操作した Delta Lake のデータを用い、Delta Lake 用の BigLake 外部テーブルを作成する手順を解説します。

出所：Delta Lake 用の BigLake 外部テーブルを作成する | BigQuery | Google Cloud

ドキュメントにて最新のメタデータをディレクトリから参照する仕様が記載されており、本記事の結果としてもメタデータの最新化の処理は不要でした。

Delta Lake テーブルにクエリを実行すると、BigQuery は接頭辞の下にあるデータを読み取ってテーブルの現在のバージョンを識別し、テーブルのメタデータとファイルを計算します。

出所：Delta Lake 用の BigLake 外部テーブルを作成する | BigQuery | Google Cloud

利用時の制限事項についてドキュメントに記載されています。

出所：Delta Lake 用の BigLake 外部テーブルを作成する | BigQuery | Google Cloud

事前準備

下記の記事を参考にして Google Colab と GCS を接続する方法を確立してください。

Google Colab の Spark にて Google Cloud Storage と連携する方法 #GoogleCloud - Qiita

手順

Delta Lake 用の BigLake 外部テーブルの作成

1. Google Colab にて Spark のバージョンを確認

!pyspark --version

2. SparkSession の定義

/content/drive/MyDrive/Colab Notebooks/bigquery_iceberg/gcs_key.jsonに GCS への接続情報を配置することを前提としています。

from google.colab import drive
drive.mount('/content/drive')

3. GCS 上に Delta Lake テーブルを作成

gcs_deltalake_path = "gs://manabianqiita/deltalake"
gcs_key_path = "/content/drive/MyDrive/Colab Notebooks/bigquery_iceberg/gcs_key.json"

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
        .appName("DeltaLakeExample")
        .config("spark.jars", "https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop3-latest.jar")
        .config("spark.jars.packages", "io.delta:delta-spark_2.12:3.2.1")
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
        .config("spark.sql.warehouse.dir", gcs_deltalake_path)
        .config("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")
        .config("google.cloud.auth.service.account.enable", "true")
        .config("spark.hadoop.google.cloud.auth.service.account.enable", "true")
        .config("spark.hadoop.google.cloud.auth.service.account.json.keyfile", gcs_key_path)
        .getOrCreate()
)
spark

4. Delta Lake テーブルの作成とデータ挿入

spark.sql("""
CREATE OR REPLACE TABLE first_table (
  id INT,
  name STRING
)
USING delta
""")

spark.sql("""
INSERT INTO
    default.first_table (id, name, status)
VALUES
    (1, 'Alice', 'active'),
    (2, 'Bob', 'active'),
    (3, 'Charlie', 'active'),
    (4, 'Daisy', 'active'),
    (5, 'Ethan', 'active');
""").show()

GCS にて Delta Lake のディレクトリが作成されていることを確認します。

5. Delta Lake 用の BigLake 外部テーブルの作成

{project_id}を実際のプロジェクト ID に置き換えて以下のコマンドを実行します。

CREATE OR REPLACE EXTERNAL TABLE `{project_id}.qiita.delta_lake_first_table`
OPTIONS(
  format="DELTA_LAKE",
  uris=["gs://manabianqiita/deltalake/first_table"]
);

CONNECTION （接続）を利用せずにテーブルを作成しておりますが、本番環境では非推奨のようです。

接続なしでも Delta Lake 外部テーブルを作成できますが、次の理由からおすすめしません。

出所：Delta Lake 用の BigLake 外部テーブルを作成する | BigQuery | Google Cloud

6. Delta Lake 用の BigLake 外部テーブルに対するクエリ実行

SELECT * FROM `qiita.delta_lake_first_table` ORDER BY id LIMIT 10;

Delta Lake テーブルのデータ操作後の Delta Lake 用の BigLake 外部テーブルの確認

1. Delta Lake テーブルに対する DELETE を実施

spark.sql("""
DELETE FROM 
    default.first_table
    WHERE
        id > 3;
""").show()

2. BigLake 外部テーブルの作成

メタデータの更新を実施することなく、最新データを取得することができます。

SELECT * FROM `qiita.delta_lake_first_table` ORDER BY id LIMIT 10;