Ce blog montrera comment charger des données d'une table Iceberg dans PyArrow ou DuckDB à l'aide de PyIceberg. Le code est accessible au public sur le référentiel Github docker-spark-iceberg qui contient une pile complète de docker-compose d'Apache Spark avec prise en charge d'Iceberg, MinIO en tant que backend de stockage et un catalogue REST en tant que catalogue. Ceci est un complément à notre vidéo PyIceberg vidéo 0.2.1. Il nécessite que docker et docker-compose soient installés.

Commençons par cloner le dépôt et afficher la pile :

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

Utilisez le cahier PyIceberg - Getting Started.ipynbpour créer le tableaunyc.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))

Chargeons quelques données :

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

Charger des données dans PyArrow

Tout d'abord, un catalogue doit être défini pour charger une table. Le docker-compose est livré avec un catalogue REST utilisé pour stocker les métadonnées. Au lieu de fournir par programme l'emplacement et les informations d'identification, vous pouvez également définir un fichier de configuration .

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

Les données contiennent des valeurs aberrantes. Filtrons les valeurs aberrantes :

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 permet aux utilisateurs d'itérer rapidement différentes parties de l'ensemble de données et de charger les données nécessaires à l'analyse.

Requête à l'aide de DuckDB

Avec PyArrow vient également le support de DuckDB. Si vous êtes un passionné de SQL comme moi, alors vous allez adorer DuckDB. La première cellule prépare la requête :

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

SELECT tip_amount
FROM df

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

Filtrons les données de la même manière qu'avec Python, mais maintenant en 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

Je recommanderais de lancer la pile et d'exécuter les requêtes pour rendre les tracés interactifs.

J'espère que vous avez appris quelque chose sur PyIcebergaujourd'hui. Si vous rencontrez quelque chose, n'hésitez pas à contacter la #pythonchaîne sur Slack ou à créer un problème sur Github .