Blog ini akan mendemonstrasikan cara memuat data dari tabel Iceberg ke PyArrow atau DuckDB menggunakan PyIceberg. Kode ini tersedia untuk umum di repositori Github docker-spark-iceberg yang berisi tumpukan lengkap docker-compose Apache Spark dengan dukungan Iceberg, MiniO sebagai backend penyimpanan, dan katalog REST sebagai katalog. Ini adalah bagian pendamping untuk video PyIceberg video 0.2.1 kami. Ini mengharuskan docker dan docker-compose diinstal.

Pertama, mari klon repositori dan buka tumpukan:

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

Gunakan notebook PyIceberg - Getting Started.ipynbuntuk membuat tabel 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))

Mari memuat beberapa data:

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

Muat data ke dalam PyArrow

Pertama, katalog perlu didefinisikan untuk memuat tabel. Docker-compose hadir dengan katalog REST yang digunakan untuk menyimpan metadata. Alih-alih menyediakan lokasi dan kredensial secara terprogram, Anda juga dapat menyetel file konfigurasi .

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

Data tersebut mengandung outlier. Mari kita menyaring outlier:

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 memungkinkan pengguna untuk dengan cepat mengulangi bagian-bagian berbeda dari kumpulan data, dan memuat data yang diperlukan untuk analisis.

Kueri menggunakan DuckDB

Dengan PyArrow juga datang dukungan dari DuckDB. Jika Anda seorang penggemar SQL seperti saya, maka Anda akan menyukai DuckDB. Sel pertama menyiapkan kueri:

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

SELECT tip_amount
FROM df

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

Mari memfilter data dengan cara yang sama seperti dengan Python, tetapi sekarang dalam 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

Saya akan merekomendasikan untuk mengaktifkan tumpukan dan menjalankan kueri untuk membuat plot menjadi interaktif.

Harap Anda telah belajar sesuatu tentang PyIceberghari ini. Jika Anda mengalami sesuatu, jangan ragu untuk menghubungi #pythonsaluran di Slack , atau membuat masalah di Github .