このブログでは、PyIceberg を使用して Iceberg テーブルから PyArrow または DuckDB にデータをロードする方法を説明します。このコードは、 docker-spark-iceberg Github リポジトリで公開されています。このリポジトリには、Iceberg をサポートする Apache Spark の完全な docker-compose スタック、ストレージ バックエンドとしての MinIO、およびカタログとしての REST カタログが含まれています。これは、PyIcebergビデオ0.2.1 ビデオの関連作品です。dockerとdocker-compose がインストールされている必要があります。

まず、リポジトリのクローンを作成してスタックを起動しましょう。

git clone https://github.com/tabular-io/docker-spark-iceberg.git
cd docker-spark-iceberg
docker-compose up

ノートブックを使用してPyIceberg - Getting Started.ipynbテーブルを作成しますnyc.taxis。

%%sql
CREATE DATABASE IF NOT EXISTS nyc;

CREATE TABLE IF NOT EXISTS nyc.taxis (
    VendorID              bigint,
    tpep_pickup_datetime  timestamp,
    tpep_dropoff_datetime timestamp,
    passenger_count       double,
    trip_distance         double,
    RatecodeID            double,
    store_and_fwd_flag    string,
    PULocationID          bigint,
    DOLocationID          bigint,
    payment_type          bigint,
    fare_amount           double,
    extra                 double,
    mta_tax               double,
    tip_amount            double,
    tolls_amount          double,
    improvement_surcharge double,
    total_amount          double,
    congestion_surcharge  double,
    airport_fee           double
)
USING iceberg
PARTITIONED BY (days(tpep_pickup_datetime))

データをロードしてみましょう:

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Jupyter").getOrCreate()

for filename in [
    "yellow_tripdata_2022-04.parquet",
    "yellow_tripdata_2022-03.parquet",
    "yellow_tripdata_2022-02.parquet",
    "yellow_tripdata_2022-01.parquet",
    "yellow_tripdata_2021-12.parquet",
]:
    df = spark.read.parquet(f"/home/iceberg/data/{filename}")
    df.write.mode("append").saveAsTable("nyc.taxis")

データを PyArrow にロードする

まず、テーブルをロードするためにカタログを定義する必要があります。docker-compose には、メタデータの保存に使用される REST カタログが付属しています。プログラムで場所と資格情報を指定する代わりに、構成ファイルを設定することもできます。

from pyiceberg.catalog import load_catalog

catalog = load_catalog('default', **{
    'uri': 'http://rest:8181',
    's3.endpoint': 'http://minio:9000',
    's3.access-key-id': 'admin',
    's3.secret-access-key': 'password',
})

tbl = catalog.load_table('nyc.taxis')

from pyiceberg.expressions import GreaterThanOrEqual

sc = tbl.scan(row_filter=GreaterThanOrEqual("tpep_pickup_datetime", "2022-01-01T00:00:00.000000+00:00"))

df = sc.to_arrow().to_pandas()

RangeIndex: 12671129 entries, 0 to 12671128
Data columns (total 19 columns):
 #   Column                 Dtype              
---  ------                 -----              
 0   VendorID               int64              
 1   tpep_pickup_datetime   datetime64[ns, UTC]
 2   tpep_dropoff_datetime  datetime64[ns, UTC]
 3   passenger_count        float64            
 4   trip_distance          float64            
 5   RatecodeID             float64            
 6   store_and_fwd_flag     object             
 7   PULocationID           int64              
 8   DOLocationID           int64              
 9   payment_type           int64              
 10  fare_amount            float64            
 11  extra                  float64            
 12  mta_tax                float64            
 13  tip_amount             float64            
 14  tolls_amount           float64            
 15  improvement_surcharge  float64            
 16  total_amount           float64            
 17  congestion_surcharge   float64            
 18  airport_fee            float64            
dtypes: datetime64[ns, UTC](2), float64(12), int64(4), object(1)
memory usage: 1.8+ GB

データには外れ値が含まれています。外れ値をフィルターで除外しましょう。

import numpy as np
from scipy import stats

stats.zscore(df['fare_amount'])

# Remove everything larger than 3 stddev
df = df[(np.abs(stats.zscore(df['fare_amount'])) < 3)]
# Remove everything below zero
df = df[df['fare_amount'] > 0]

PyIceberg を使用すると、ユーザーはデータセットのさまざまな部分を迅速に繰り返し、分析に必要なデータを読み込むことができます。

DuckDBを使用したクエリ

PyArrow では、DuckDB によるサポートも提供されます。あなたが私のような SQL 愛好家であれば、DuckDB を気に入るはずです。最初のセルはクエリを準備します。

%%sql --save tip-amount --no-execute

SELECT tip_amount
FROM df

%sqlplot histogram --table df --column tip_amount --bins 22 --with tip-amount

Python と同様の方法でデータをフィルタリングしてみましょう。ただし、今度は SQL を使用します。

%%sql --save tip-amount-filtered --no-execute

WITH tip_amount_stddev AS (
    SELECT STDDEV_POP(tip_amount) AS tip_amount_stddev
    FROM df
)

SELECT tip_amount
FROM df, tip_amount_stddev
WHERE tip_amount > 0
  AND tip_amount < tip_amount_stddev * 3

%sqlplot histogram --table tip-amount-filtered --column tip_amount --bins 50 --with tip-amount-filtered

スタックを起動してクエリを実行して、プロットをインタラクティブにすることをお勧めします。

今日のことで何かを学んでいただければ幸いですPyIceberg。何か問題が発生した場合は、遠慮なくSlack#pythonのチャンネルに連絡するか、Github で問題を作成してください。