Apache Iceberg¶

from pyspark.sql import SparkSession

ICEBERG_VERSION = "1.6.1"
SPARK_VERSION   = "3.5"
SCALA_VERSION   = "2.12"
JAR = (
    f"org.apache.iceberg:iceberg-spark-runtime"
    f"-{SPARK_VERSION}_{SCALA_VERSION}:{ICEBERG_VERSION}"
)

spark = SparkSession.builder \
    .appName("ApacheIceberg") \
    .master("local[*]") \
    .config("spark.jars.packages", JAR) \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.local",
            "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.local.type", "hadoop") \
    .config("spark.sql.catalog.local.warehouse", "warehouse/iceberg") \
    .config("spark.sql.defaultCatalog", "local") \
    .getOrCreate()

spark.sparkContext.setLogLevel("ERROR")

# Criar namespace (banco de dados)
spark.sql("CREATE NAMESPACE IF NOT EXISTS local.techstore")

# Criar tabela Iceberg via SQL
spark.sql("""
    CREATE TABLE IF NOT EXISTS local.techstore.clientes (
        cliente_id   INT    NOT NULL,
        nome         STRING NOT NULL,
        email        STRING NOT NULL,
        cidade       STRING,
        estado       STRING
    )
    USING iceberg
    PARTITIONED BY (estado)
""")

# Via DataFrame
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

schema = StructType([
    StructField("cliente_id", IntegerType(), False),
    StructField("nome",       StringType(),  False),
    StructField("email",      StringType(),  False),
    StructField("cidade",     StringType(),  True),
    StructField("estado",     StringType(),  True),
])

dados = [
    (1, "Alice Souza",    "alice@email.com",  "Criciúma",     "SC"),
    (2, "Bruno Oliveira", "bruno@email.com",  "Florianópolis","SC"),
    (3, "Carla Mendes",   "carla@email.com",  "São Paulo",    "SP"),
    (4, "Diego Alves",    "diego@email.com",  "Curitiba",     "PR"),
]

df = spark.createDataFrame(dados, schema=schema)

df.writeTo("local.techstore.clientes") \
  .append()

# Via SQL
spark.sql("""
    INSERT INTO local.techstore.clientes VALUES
    (5, 'Eduardo Lima', 'edu@email.com', 'Porto Alegre', 'RS')
""")

# Atualizar cidade de um cliente
spark.sql("""
    UPDATE local.techstore.clientes
    SET cidade = 'Tubarão'
    WHERE cliente_id = 1
""")

# Verificar resultado
spark.sql("""
    SELECT * FROM local.techstore.clientes
    WHERE cliente_id = 1
""").show()

# Remover cliente específico
spark.sql("""
    DELETE FROM local.techstore.clientes
    WHERE cliente_id = 3
""")

# Verificar tabela após remoção
spark.sql("SELECT * FROM local.techstore.clientes ORDER BY cliente_id").show()

# Criar tabela temporária com dados novos/atualizados
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW novos_clientes AS
    SELECT * FROM VALUES
        (2, 'Bruno O. Silva', 'bruno@email.com', 'Joinville', 'SC'),
        (6, 'Fernanda Rocha', 'fe@email.com',    'Salvador',  'BA')
    AS t(cliente_id, nome, email, cidade, estado)
""")

# MERGE INTO
spark.sql("""
    MERGE INTO local.techstore.clientes AS destino
    USING novos_clientes AS fonte
    ON destino.cliente_id = fonte.cliente_id
    WHEN MATCHED THEN
        UPDATE SET
            destino.nome   = fonte.nome,
            destino.cidade = fonte.cidade,
            destino.estado = fonte.estado
    WHEN NOT MATCHED THEN
        INSERT (cliente_id, nome, email, cidade, estado)
        VALUES (fonte.cliente_id, fonte.nome, fonte.email,
                fonte.cidade, fonte.estado)
""")

# Listar todos os snapshots da tabela
spark.sql("""
    SELECT snapshot_id, committed_at, operation, summary
    FROM local.techstore.clientes.snapshots
    ORDER BY committed_at
""").show(truncate=False)

# Ler dados de um snapshot específico
spark.sql("""
    SELECT * FROM local.techstore.clientes
    FOR SYSTEM_VERSION AS OF <snapshot_id>
""").show()

# Ler dados de um timestamp específico
spark.sql("""
    SELECT * FROM local.techstore.clientes
    FOR SYSTEM_TIME AS OF '2025-01-15 10:00:00'
""").show()

# Via DataFrame API
df_historico = spark.read \
    .option("snapshot-id", "<snapshot_id>") \
    .table("local.techstore.clientes")

# Criar tabela particionada por 'estado'
spark.sql("""
    CREATE TABLE local.techstore.pedidos (
        pedido_id   INT,
        cliente_id  INT,
        data_pedido DATE,
        status      STRING,
        valor_total DOUBLE
    )
    USING iceberg
    PARTITIONED BY (estado_cliente STRING, months(data_pedido))
""")

# Adicionar nova partição SEM reescrever dados antigos
spark.sql("""
    ALTER TABLE local.techstore.pedidos
    ADD PARTITION FIELD days(data_pedido)
""")

# Remover partição antiga
spark.sql("""
    ALTER TABLE local.techstore.pedidos
    DROP PARTITION FIELD months(data_pedido)
""")

# Adicionar coluna
spark.sql("""
    ALTER TABLE local.techstore.clientes
    ADD COLUMN telefone STRING AFTER email
""")

# Renomear coluna
spark.sql("""
    ALTER TABLE local.techstore.clientes
    RENAME COLUMN cidade TO municipio
""")

# Alterar tipo de coluna (widening apenas: int → long → float → double → decimal)
spark.sql("""
    ALTER TABLE local.techstore.clientes
    ALTER COLUMN cliente_id TYPE BIGINT
""")

# Remover coluna
spark.sql("""
    ALTER TABLE local.techstore.clientes
    DROP COLUMN telefone
""")

# Compactar arquivos pequenos (bin-packing)
spark.sql("""
    CALL local.system.rewrite_data_files(
        table => 'local.techstore.clientes',
        options => map('target-file-size-bytes', '134217728')
    )
""")

# Expirar snapshots antigos (manter últimas 24 horas)
spark.sql("""
    CALL local.system.expire_snapshots(
        table => 'local.techstore.clientes',
        older_than => TIMESTAMP '2025-01-01 00:00:00',
        retain_last => 5
    )
""")

# Remover arquivos órfãos
spark.sql("""
    CALL local.system.remove_orphan_files(
        table => 'local.techstore.clientes'
    )
""")