Apache Iceberg

-- Criar tabela Iceberg
CREATE TABLE my_catalog.db.events (
    id bigint,
    event_time timestamp,
    user_id string,
    event_type string
) USING iceberg
PARTITIONED BY (days(event_time))

-- Inserir dados
INSERT INTO my_catalog.db.events VALUES 
(1, '2024-01-01 10:00:00', 'user1', 'login'),
(2, '2024-01-01 11:00:00', 'user2', 'purchase')

-- Time travel query
SELECT * FROM my_catalog.db.events 
TIMESTAMP AS OF '2024-01-01 10:30:00'

-- Schema evolution
ALTER TABLE my_catalog.db.events 
ADD COLUMN session_id string

from pyiceberg.catalog import load_catalog
import pandas as pd

# Carregar catalog
catalog = load_catalog("my_catalog")

# Criar tabela
schema = pa.schema([
    pa.field("id", pa.int64()),
    pa.field("event_time", pa.timestamp('us')),
    pa.field("user_id", pa.string()),
    pa.field("event_type", pa.string())
])

table = catalog.create_table(
    "db.events",
    schema=schema,
    partition_spec=PartitionSpec(
        PartitionField(source_id=2, field_id=1000, transform=DayTransform(), name="event_time_day")
    )
)

# Inserir dados
df = pd.DataFrame({
    'id': [1, 2, 3],
    'event_time': pd.to_datetime(['2024-01-01 10:00:00', '2024-01-01 11:00:00', '2024-01-01 12:00:00']),
    'user_id': ['user1', 'user2', 'user3'],
    'event_type': ['login', 'purchase', 'logout']
})

table.append(df)

# Ler dados
scan = table.scan()
df_result = scan.to_pandas()

-- Configurar catalog Iceberg
CREATE CATALOG iceberg_catalog WITH (
  'type'='iceberg',
  'catalog-type'='hive',
  'uri'='thrift://localhost:9083',
  'warehouse'='hdfs://namenode:port/path/to/warehouse'
);

USE CATALOG iceberg_catalog;

-- Criar tabela para streaming
CREATE TABLE events (
    id BIGINT,
    event_time TIMESTAMP(3),
    user_id STRING,
    event_type STRING,
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) PARTITIONED BY (event_type);

-- Streaming insert
INSERT INTO events 
SELECT id, event_time, user_id, event_type
FROM kafka_source;