Questo blog dimostrerà come caricare i dati da una tabella Iceberg in PyArrow o DuckDB utilizzando PyIceberg. Il codice è pubblicamente disponibile nel repository Github docker-spark-iceberg che contiene uno stack completo di Docker Compose di Apache Spark con supporto Iceberg, MinIO come back-end di archiviazione e un catalogo REST come catalogo. Questo è un complemento del nostro video PyIceberg 0.2.1 . Richiede che docker e docker-compose siano installati.

Per prima cosa cloniamo il repository e apriamo lo stack:

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

Usa il taccuino PyIceberg - Getting Started.ipynbper creare la tabella 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))

Carichiamo alcuni dati:

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

Carica i dati in PyArrow

Innanzitutto, è necessario definire un catalogo per caricare una tabella. La finestra mobile-compose viene fornita con un catalogo REST utilizzato per archiviare i metadati. Invece di fornire a livello di codice la posizione e le credenziali, puoi anche impostare un file di configurazione .

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

I dati contengono valori anomali. Filtriamo gli 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 consente agli utenti di iterare rapidamente diverse parti del set di dati e caricare i dati necessari per l'analisi.

Interrogare utilizzando DuckDB

Con PyArrow arriva anche il supporto di DuckDB. Se sei un appassionato di SQL come me, adorerai DuckDB. La prima cella prepara la query:

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

SELECT tip_amount
FROM df

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

Filtriamo i dati in modo simile a Python, ma ora in 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

Consiglierei di accendere lo stack ed eseguire le query per rendere interattivi i grafici.

Spero che tu abbia imparato qualcosa su PyIcebergoggi. Se ti imbatti in qualcosa, non esitare a contattare il #pythoncanale su Slack o a creare un problema su Github .