DatabricksでS3に外部テーブルとして作成したDeltaテーブルにローカル環境のPythonからアクセスしてみた

sample-data-delta/
├── _delta_log/
│   ├── 00000000000000000000.crc
│   ├── 00000000000000000000.json
│   ├── _autostats/
│   │   └── 00000000000000000000.0000000000.stats.json
│   └── _commits/
├── yyyymmdd=20250327/
│   └── part-00001-71449f89-62b4-4d5e-95c9-77df7bd159e6.c000.snappy.parquet
└── yyyymmdd=20250328/
    └── part-00000-6b34c5c2-962b-483b-9310-d11a533bf187.c000.snappy.parquet

$ pip install duckdb

import duckdb

# DuckDB接続を作成
con = duckdb.connect()

create_secret_query="""
CREATE OR REPLACE SECRET s3_secret (
    TYPE s3,
    PROVIDER credential_chain,
    CHAIN env
);
"""

# DeltaテーブルがあるS3のパス
s3_path = "s3://uehara-databricks-test/sample-data-delta/"

try:
    # シークレットの作成
    con.execute(create_secret_query)

    # S3上のDeltaテーブルを読み込む
    query = f"SELECT * FROM delta_scan('{s3_path}')"
    print(f"SQL: {query}")
    # クエリを実行してpandas DataFrameに変換
    df = con.execute(query).fetchdf()

    print(df.to_string(index=False))
except Exception as e:
    print(f"Error accessing Delta table at {s3_path}: {e}")
finally:
    # 接続を閉じる
    con.close()

#!/usr/bin/env bash
set -euo pipefail

PROFILE=$1

unset AWS_PROFILE
export AWS_SDK_LOAD_CONFIG=true

ROLE_ARN=$(aws --profile ${PROFILE} configure get role_arn)
STS_RESULT=$(aws --profile ${PROFILE} sts assume-role --role-arn ${ROLE_ARN} --role-session-name sls-deploy-session --duration-seconds 3600)

export AWS_ACCESS_KEY_ID=$(echo ${STS_RESULT} | jq -r '.Credentials.AccessKeyId')
export AWS_SECRET_ACCESS_KEY=$(echo ${STS_RESULT} | jq -r '.Credentials.SecretAccessKey')
export AWS_SESSION_TOKEN=$(echo ${STS_RESULT} | jq -r '.Credentials.SessionToken')
export AWS_REGION='ap-northeast-1'

python sample_1.py

$ bash get_secret_and_execute_py.sh <YOUR-AWS-PROFILE>

The table has set these reader features: {'deletionVectors'} but these are not yet supported by the deltalake reader.

$ pip install databricks-sdk

$ export DATABRICKS_HOST=dbc-xxxxxxxx-xxxx.cloud.databricks.com

$ export DATABRICKS_TOKEN=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

import time

import pandas as pd
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.sql import StatementState

# SQL warehouseのID
WAREHOUSE_ID = "<YOUR WAREHOUSE ID>"

# Databricks SDKクライアント初期化
w = WorkspaceClient()

try:
    # SQLの実行
    sql_query = "SELECT * FROM workspace.default.sample_data_delta"
    print(f"SQL: {sql_query}")
    resp = w.statement_execution.execute_statement(
        statement=sql_query,
        warehouse_id=WAREHOUSE_ID,
        wait_timeout='50s' # タイムアウト設定 (例: 50秒)、 '0s' で非同期実行
    )

    # 実行ステータスの確認
    statement_id = resp.statement_id
    print(f"Statement ID: {statement_id}")
    print(f"Current state: {resp.status.state}")

    if resp.status.state in (StatementState.SUCCEEDED, StatementState.FAILED, StatementState.CANCELED, StatementState.CLOSED):
        print("Statement execution finished.")
    else:
        # 通常wait_timeoutを設定すれば完了を待つが、指定時間内で処理が完了せずポーリングが必要な場合
        # （非同期実行でも同様の処理となる）
        while resp.status.state in (StatementState.PENDING, StatementState.RUNNING):
            print("Waiting for statement execution to complete...")
            time.sleep(5)
            resp = w.statement_execution.get_statement(statement_id=statement_id)
            print(f"Current state: {resp.status.state}")

    # 結果の取得
    if resp.status.state == StatementState.SUCCEEDED:
        print("Fetching results...")

        # ステートメント情報を取得して列名を抽出
        statement_info = w.statement_execution.get_statement(statement_id=statement_id)
        column_names = [col.name for col in statement_info.manifest.schema.columns]

        # 最初のチャンクのデータを取得（すでにstatement_infoに含まれている）
        data = statement_info.result.data_array
        df = pd.DataFrame(data, columns=column_names)

        # 次のチャンクのデータがあるのであれば取得
        next_chunk_index = statement_info.result.next_chunk_index
        while next_chunk_index is not None:
            print(f"Fetching chunk {next_chunk_index}...")
            chunk_result = w.statement_execution.get_statement_result_chunk_n(
                statement_id=statement_id,
                chunk_index=next_chunk_index
            )
            if chunk_result.data_array:
                # 追加データをDataFrameに追加
                next_df = pd.DataFrame(chunk_result.data_array, columns=column_names)
                df = pd.concat([df, next_df], ignore_index=True)
            next_chunk_index = chunk_result.next_chunk_index

        # 結果の表示
        print(df.to_string(index=False))

    elif resp.status.state in (StatementState.FAILED, StatementState.CANCELED, StatementState.CLOSED):
        print(f"Statement execution did not succeed: {resp.status.state}")
        if resp.status.error:
            print(f"Error message: {resp.status.error.message}")

except Exception as e:
    print(f"An error occurred: {e}")