Iceberg

Permite a configuração de uma tabela Apache Iceberg para escrita de dados.

Parâmetros

Parâmetros aceitos para esse tipo de saída:

Campo	Descrição	Valor
iceberg.catalog	Configuração do catálogo Iceberg	Tipo: object Obrigatório
iceberg.catalog.name	Nome do catálogo	Tipo: string Obrigatório
iceberg.catalog.type	Tipo do catálogo	Tipo: enum[rest] Valor padrão: rest
iceberg.catalog.uri	URI de conexão com o catálogo REST	Tipo: string Obrigatório
iceberg.catalog.warehouse	Caminho do warehouse (ex: gs://bucket/warehouse)	Tipo: string Obrigatório
iceberg.catalog.storage	Configuração de armazenamento	Tipo: object Obrigatório
iceberg.catalog.storage.type	Tipo de armazenamento	Tipo: enum[gcs] Valor padrão: gcs
iceberg.catalog.storage.gcs	Configuração de credenciais GCS	Tipo: object Obrigatório se storage.type = gcs
iceberg.catalog.storage.gcs.credential	Credenciais de acesso ao GCS	Tipo: object Obrigatório
iceberg.catalog.storage.gcs.credential.type	O tipo de conta	Tipo: string Valor padrão: service_account
iceberg.catalog.storage.gcs.credential.project_id	O ID do seu projeto no Google Cloud	Tipo: string Obrigatório
iceberg.catalog.storage.gcs.credential.private_key_id	O ID da sua chave privada	Tipo: string Obrigatório
iceberg.catalog.storage.gcs.credential.private_key	A sua chave privada	Tipo: string Obrigatório
iceberg.catalog.storage.gcs.credential.client_email	O endereço de email da sua conta de serviço	Tipo: string Obrigatório
iceberg.catalog.storage.gcs.credential.client_id	O ID da sua conta de serviço	Tipo: string Obrigatório
iceberg.catalog.storage.gcs.credential.auth_uri	O URI para autorização OAuth2	Tipo: string Valor padrão: https://accounts.google.com/o/oauth2/auth
iceberg.catalog.storage.gcs.credential.token_uri	O URI que fornece os tokens OAuth2	Tipo: string Valor padrão: https://oauth2.googleapis.com/token
iceberg.catalog.storage.gcs.credential.auth_provider_x509_cert_url	O URL do certificado X.509 do provedor de autenticação	Tipo: string Valor padrão: https://www.googleapis.com/oauth2/v1/certs
iceberg.catalog.storage.gcs.credential.client_x509_cert_url	O URL do certificado X.509 da conta de serviço	Tipo: string Obrigatório
iceberg.table_identifier	Identificador da tabela (formato: namespace.tabela)	Tipo: string Obrigatório
iceberg.write_mode	Modo de escrita	Tipo: enum[append] Valor padrão: append
iceberg.max_in_flight	Número máximo de escritas simultâneas	Tipo: int Valor padrão: 1
iceberg.batching	Configuração de agrupamento de mensagens	Tipo: object Opcional
iceberg.batching.count	Número de mensagens por lote	Tipo: int[1-100000] Valor padrão: 1000
iceberg.batching.period	Período máximo de espera por lote	Tipo: string Valor padrão: 5s
iceberg.create_table	Configuração para criação automática da tabela	Tipo: object Opcional
iceberg.create_table.schema	Definição do schema da tabela	Tipo: []object Opcional
iceberg.create_table.schema[].name	Nome do campo	Tipo: string Obrigatório
iceberg.create_table.schema[].type	Tipo do campo	Tipo: enum[boolean int long float double string binary date time timestamp timestamptz] Obrigatório
iceberg.create_table.schema[].required	Se o campo é obrigatório	Tipo: boolean Valor padrão: false
iceberg.create_table.schema[].write_default	Valor padrão para escrita	Tipo: any Opcional
iceberg.create_table.partition_spec	Especificação de particionamento	Tipo: []object Opcional
iceberg.create_table.partition_spec[].source_id	ID do campo fonte	Tipo: string Obrigatório
iceberg.create_table.partition_spec[].name	Nome da partição	Tipo: string Obrigatório
iceberg.create_table.partition_spec[].transform	Transformação aplicada (ex: day, month, year, bucket, truncate)	Tipo: string Obrigatório
iceberg.create_table.sort_order	Especificação de ordenação	Tipo: []object Opcional
iceberg.create_table.sort_order[].source_id	ID do campo fonte	Tipo: string Obrigatório
iceberg.create_table.sort_order[].transform	Transformação aplicada	Tipo: enum[identity bucket truncate] Valor padrão: identity
iceberg.create_table.sort_order[].direction	Direção da ordenação	Tipo: enum[asc desc] Opcional
iceberg.create_table.sort_order[].null_order	Ordenação de valores nulos	Tipo: enum[nulls-first nulls-last] Opcional

Requisição

Exemplo Básico (tabela existente)

curl --location '<HOST>/v1/configs' \
--header 'Content-Type: application/json' \
--header 'Authorization: <ACCESS-TOKEN>' \
--data '{
    "input": {...},
    "output": {
        "type": "ICEBERG",
        "iceberg": {
            "catalog": {
                "name": "my_catalog",
                "type": "rest",
                "uri": "https://iceberg-catalog.example.com",
                "warehouse": "gs://my-bucket/warehouse",
                "storage": {
                    "type": "gcs",
                    "gcs": {
                        "credential": {
                            "type": "service_account",
                            "project_id": "PROJECT_ID",
                            "private_key_id": "PRIVATE_KEY_ID",
                            "private_key": "PRIVATE_KEY",
                            "client_email": "CLIENT_EMAIL",
                            "client_id": "CLIENT_ID",
                            "auth_uri": "https://accounts.google.com/o/oauth2/auth",
                            "token_uri": "https://oauth2.googleapis.com/token",
                            "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
                            "client_x509_cert_url": "CLIENT_X509_CERT_URL"
                        }
                    }
                }
            },
            "table_identifier": "db.events",
            "write_mode": "append",
            "batching": {
                "count": 5000,
                "period": "10s"
            }
        }
    }
}'

Exemplo com Criação de Tabela

curl --location '<HOST>/v1/configs' \
--header 'Content-Type: application/json' \
--header 'Authorization: <ACCESS-TOKEN>' \
--data '{
    "input": {...},
    "output": {
        "type": "ICEBERG",
        "iceberg": {
            "catalog": {
                "name": "my_catalog",
                "type": "rest",
                "uri": "https://iceberg-catalog.example.com",
                "warehouse": "gs://my-bucket/warehouse",
                "storage": {
                    "type": "gcs",
                    "gcs": {
                        "credential": {
                            "type": "service_account",
                            "project_id": "PROJECT_ID",
                            "private_key_id": "PRIVATE_KEY_ID",
                            "private_key": "PRIVATE_KEY",
                            "client_email": "CLIENT_EMAIL",
                            "client_id": "CLIENT_ID"
                        }
                    }
                }
            },
            "table_identifier": "db.events",
            "write_mode": "append",
            "create_table": {
                "schema": [
                    {"name": "id", "type": "long", "required": true},
                    {"name": "event_type", "type": "string", "required": true},
                    {"name": "payload", "type": "string"},
                    {"name": "created_at", "type": "timestamptz", "required": true}
                ],
                "partition_spec": [
                    {"source_id": "created_at", "name": "day", "transform": "day"}
                ],
                "sort_order": [
                    {"source_id": "created_at", "transform": "identity", "direction": "desc"}
                ]
            },
            "batching": {
                "count": 10000,
                "period": "30s"
            }
        }
    }
}'

Resposta

A resposta desse endpoint será no seguinte formato:

{
  "id": "FIREHOSE-ID",
  "output": {
    "type": "ICEBERG",
    "iceberg": {
      "catalog": {
        "name": "my_catalog",
        "type": "rest",
        "uri": "https://iceberg-catalog.example.com",
        "warehouse": "gs://my-bucket/warehouse",
        "storage": {
          "type": "gcs",
          "gcs": {
            "credential": {
              "type": "service_account",
              "project_id": "PROJECT_ID",
              "private_key_id": "*",
              "private_key": "*",
              "client_email": "*",
              "client_id": "*",
              "auth_uri": "https://accounts.google.com/o/oauth2/auth",
              "token_uri": "https://oauth2.googleapis.com/token",
              "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
              "client_x509_cert_url": "*"
            }
          }
        }
      },
      "table_identifier": "db.events",
      "write_mode": "append",
      "batching": {
        "count": 5000,
        "period": "10s"
      }
    }
  }
}

O campo table_identifier deve estar no formato namespace.tabela. Se a tabela não existir e create_table for configurado, ela será criada automaticamente com o schema especificado.

O modo de escrita atual suporta apenas append. Outros modos como overwrite podem ser adicionados em versões futuras.

A configuração de batching é recomendada para otimizar a performance de escrita, agrupando múltiplas mensagens em um único commit.