Este blog demonstrará como carregar dados de uma tabela Iceberg em PyArrow ou DuckDB usando PyIceberg. O código está disponível publicamente no repositório Github docker-spark-iceberg , que contém uma pilha completa de composição docker do Apache Spark com suporte a Iceberg, MinIO como um back-end de armazenamento e um catálogo REST como um catálogo. Este é um complemento do nosso vídeo PyIceberg 0.2.1. Requer que o docker e o docker-compose estejam instalados.

Primeiro, vamos clonar o repositório e abrir a pilha:

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

Use o notebook PyIceberg - Getting Started.ipynbpara criar a tabela 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))

Vamos carregar alguns dados:

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

Carregar dados no PyArrow

Primeiro, um catálogo precisa ser definido para carregar uma tabela. O docker-compose vem com um catálogo REST usado para armazenar os metadados. Em vez de fornecer programaticamente o local e as credenciais, você também pode definir um arquivo de configuração .

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

Os dados contêm outliers. Vamos filtrar os outliers:

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]

O PyIceberg permite que os usuários iterem rapidamente diferentes partes do conjunto de dados e carreguem os dados necessários para a análise.

Consultar usando DuckDB

Com o PyArrow também vem o suporte do DuckDB. Se você é um entusiasta do SQL como eu, vai adorar o DuckDB. A primeira célula prepara a consulta:

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

SELECT tip_amount
FROM df

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

Vamos filtrar os dados de forma semelhante ao Python, mas agora em 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

Eu recomendaria ativar a pilha e executar as consultas para tornar os gráficos interativos.

Espero que você tenha aprendido algo sobre PyIceberghoje. Se você se deparar com alguma coisa, não hesite em entrar em contato no #pythoncanal no Slack ou criar um problema no Github .