Bu blog, PyIceberg kullanılarak bir Iceberg tablosundan PyArrow veya DuckDB'ye nasıl veri yükleneceğini gösterecek. Kod, Iceberg destekli tam bir docker-compose Apache Spark yığını, depolama arka ucu olarak MinIO ve katalog olarak bir REST kataloğu içeren docker-spark-iceberg Github deposunda herkesin kullanımına açıktır. Bu, PyIceberg video 0.2.1 videomuza eşlik eden bir parçadır . Docker ve docker-compose'un kurulu olmasını gerektirir .

İlk olarak, depoyu klonlayalım ve yığını getirelim:

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

PyIceberg - Getting Started.ipynbTabloyu oluşturmak için not defterini kullanın 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))

Biraz veri yükleyelim:

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

Verileri PyArrow'a yükleyin

İlk olarak, bir tablo yüklemek için bir kataloğun tanımlanması gerekir. docker-compose, meta verileri depolamak için kullanılan bir REST kataloğu ile birlikte gelir. Konum ve kimlik bilgilerini program aracılığıyla sağlamak yerine, bir yapılandırma dosyası da ayarlayabilirsiniz .

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

Veriler aykırı değerler içeriyor. Aykırı değerleri filtreleyelim:

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, kullanıcıların veri kümesinin farklı bölümlerini hızla yinelemesine ve analiz için gereken verileri yüklemesine olanak tanır.

DuckDB kullanarak sorgulama

PyArrow ile DuckDB'den de destek geliyor. Benim gibi bir SQL meraklısıysanız, DuckDB'ye bayılacaksınız. İlk hücre sorguyu hazırlar:

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

SELECT tip_amount
FROM df

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

Verileri Python ile benzer bir şekilde filtreleyelim, ancak şimdi SQL'de:

%%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

Çizimleri etkileşimli hale getirmek için yığını çalıştırmanızı ve sorguları çalıştırmanızı tavsiye ederim.

Umarım bugün hakkında bir şeyler öğrenmişsindir PyIceberg. Herhangi bir şeyle karşılaşırsanız, lütfen Slack'teki#python kanala ulaşmaktan veya Github'da bir sorun oluşturmaktan çekinmeyin .