Pular para o conteúdo principal
Última atualização em

Use o Arrow Flight com o Zerobus Ingest

info

Beta

A ingestão via Arrow Flight está em Beta.

A ingestão via Arrow Flight permite enviar dados do Apache Arrow RecordBatch diretamente para o Zerobus Ingest em vez de converter cada linha para JSON ou Protocol Buffers (protobuf) primeiro. É uma terceira opção de formato de registro nos SDKs do Zerobus, juntamente com JSON e protobuf, e funciona sobre a mesma conexão gRPC. Ele usa o mesmo Endpoint do Zerobus, o mesmo fluxo de OAuth e a mesma convenção de cabeçalho x-databricks-zerobus-table-name. O protocolo de rede é o Arrow Flight DoPut, que transporta mensagens IPC do Arrow via gRPC.

Quando usar o Arrow Flight​

Arrow Flight é a melhor opção nos seguintes cenários:

  • Seu aplicativo já produz dados Arrow, como pyarrow.Table ou pyarrow.RecordBatch (Python), arrow_array::RecordBatch dos crates arrow-rs (Rust) ou VectorSchemaRoot (Java). Bibliotecas de DataFrame criadas sobre Arrow, como Polars ou DataFusion, encaixam-se naturalmente neste caminho.
  • Você ingere linhas em lotes, em vez de enviar um registro por vez.
  • Seu esquema é amplo, envolve muitos dados numéricos ou é orientado para análises, onde a serialização linha por linha adiciona uma sobrecarga de CPU considerável.
  • Você está criando coletores ou gateways que agregam dados por um curto período e, em seguida, os enviam como lotes em formato de coluna única.

O Arrow Flight geralmente não é a melhor escolha para tráfego esparso, linha por linha. Nesses casos, JSON ou protobuf via caminho gRPC do SDK são normalmente mais simples. Consulte Escolha uma interface.

Como funciona o modelo de ingestão​

Com a ingestão do Arrow Flight, uma transmissão grava em uma tabela de destino. Para ingerir dados, siga esta sequência:

  1. Defina um esquema Arrow que corresponda ao esquema da tabela Delta de destino.
  2. Abra uma transmissão Zerobus Arrow para essa mesa.
  3. Enviar cargas úteis RecordBatch (ou Table).
  4. Aguarde o último deslocamento ou chame flush() para confirmar a durabilidade.
  5. Feche a transmissão.

Se você usar um SDK do Zerobus, o SDK cuidará dos detalhes de baixo nível da conexão do Arrow Flight para você. Ele serializa seus dados Arrow para o formato IPC e divide um lote grande em mensagens de transporte menores, as quais o servidor reconhece individualmente.

O Arrow Flight não oferece durabilidade de tudo ou nada para todo o lote lógico. Um lote Arrow pode ser muito grande e, como o SDK o divide em mensagens de transporte separadas que são confirmadas à medida que são recebidas, um lote grande pode ser parcialmente durável se ocorrer uma falha durante o processo. Isso é diferente dos lotes JSON e protobuf, que fazem commit atomicamente e são limitados pelo tamanho de mensagem de 10 MB. Consulte Os lotes do Arrow Flight são a exceção.

A abstração de deslocamento lógico ainda se mantém sobre esse particionamento. ingest_batch() retorna um único deslocamento lógico para o lote que você enviou, e wait_for_offset() nesse deslocamento só é concluído após cada mensagem de transporte que compõe o lote ter sido confirmada. (Os nomes dos métodos são do SDK do Python; outros SDKs expõem métodos equivalentes, como ingestBatch e waitForOffset em Java.)

Assim como na regra de esquema Protobuf, o esquema que você passa para a transmissão deve se ajustar à tabela Delta de destino: ele deve conter, no mínimo, todas as colunas não anuláveis. Seu esquema pode omitir colunas anuláveis que existem na tabela Delta (isso é tratado como uma alteração de esquema não disruptiva), mas qualquer outra incompatibilidade é rejeitada. O tipo de cada campo Arrow deve ser compatível com sua coluna Delta. Para os tipos Delta compatíveis, consulte Tipos de dados compatíveis.

Como o SDK divide grandes lotes em mensagens de transporte, um lote Arrow não está sujeito ao limite de tamanho de mensagem de 10 MB da mesma forma que um lote JSON ou Protocol Buffers. Como o Arrow Flight é executado sobre o mesmo transporte gRPC, as mesmas características de throughput, latência e cota se aplicam, todas as quais escalam para atender a cargas de trabalho maiores. Veja cotas do Zerobus Ingest.

Escreva para um cliente​

Os exemplos abaixo abrem uma transmissão do Arrow Flight na mesma tabela air_quality usada nos exemplos de Usar o Zerobus Ingest. Eles são mostrados em Python e Rust por brevidade, mas o mesmo construtor, opções de configuração e sequência de chamadas estão disponíveis em todos os SDKs do Zerobus. Adapte a sintaxe para sua linguagem e consulte o repository do SDK para tipos de Arrow específicos da linguagem.

O SDK Python aceita um pyarrow.Schema no momento da criação da transmissão e um pyarrow.RecordBatch ou pyarrow.Table para cada chamada de ingestão.

Bash
pip install "databricks-zerobus-ingest-sdk[arrow]" pyarrow
Python
import pyarrow as pa

from zerobus.sdk.sync import ZerobusSdk

# See "Get your workspace URL and Zerobus Ingest endpoint" in zerobus-ingest.md.
SERVER_ENDPOINT = "https://1234567890123456.zerobus.us-west-2.cloud.databricks.com"
DATABRICKS_WORKSPACE_URL = "https://dbc-a1b2c3d4-e5f6.cloud.databricks.com"
TABLE_NAME = "main.default.air_quality"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"

schema = pa.schema(
    [
        ("device_name", pa.large_utf8()),
        ("temp", pa.int32()),
        ("humidity", pa.int64()),
    ]
)

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

stream = sdk.create_arrow_stream(TABLE_NAME, schema, CLIENT_ID, CLIENT_SECRET)

row_count = 1_000
batch = pa.record_batch(
    {
        "device_name": [f"sensor-{i}" for i in range(row_count)],
        "temp": [20 + (i % 5) for i in range(row_count)],
        "humidity": [55 + (i % 10) for i in range(row_count)],
    },
    schema=schema,
)

try:
    offset = stream.ingest_batch(batch)

    # Optional: block until the batch is durably written
    stream.wait_for_offset(offset)
finally:
    stream.close()

stream.ingest_batch() também aceita um pyarrow.Table. O SDK o converte internamente para um único RecordBatch antes do envio. Cada chamada retorna um deslocamento lógico. O bloqueio no deslocamento é opcional. Para saber quando aguardar e como funciona o reconhecimento, veja Message blocking and acknowledgment.

Ingestão de colunas VARIANT​

O Apache Arrow não possui um tipo VARIANT nativo. Para realizar a ingestão em uma coluna VARIANT via Arrow Flight, crie os campos metadata e value de suporte da coluna como uma estrutura (struct) de duas colunas LargeBinary e, em seguida, inclua essa estrutura no seu RecordBatch. Por meio dos SDKs gRPC e REST, você passa um valor Variant como uma **strings** codificada em JSON. Consulte Tipos de dados suportados.

O exemplo de Rust a seguir cria uma coluna de struct VARIANT a partir de linhas JSON e a ingere:

Rust
fn variant_struct(json_rows: &[&str]) -> ArrayRef {
    let mut metas: Vec<Vec<u8>> = Vec::new();
    let mut vals: Vec<Vec<u8>> = Vec::new();
    for json in json_rows {
        let mut vb = VariantBuilder::new();
        vb.append_json(json).expect("invalid JSON for variant");
        let (metadata, value) = vb.finish();
        metas.push(metadata);
        vals.push(value);
    }
    let fields = Fields::from(vec![
        Field::new("metadata", DataType::LargeBinary, false),
        Field::new("value", DataType::LargeBinary, false),
    ]);
    let meta_arr = Arc::new(LargeBinaryArray::from_iter_values(metas)) as ArrayRef;
    let val_arr = Arc::new(LargeBinaryArray::from_iter_values(vals)) as ArrayRef;
    Arc::new(StructArray::try_new(fields, vec![meta_arr, val_arr], None).expect("variant struct"))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client_id = std::env::var("DATABRICKS_CLIENT_ID")?;
    let client_secret = std::env::var("DATABRICKS_CLIENT_SECRET")?;

    let variant_type = DataType::Struct(Fields::from(vec![
        Field::new("metadata", DataType::LargeBinary, false),
        Field::new("value", DataType::LargeBinary, false),
    ]));
    let schema = Arc::new(ArrowSchema::new(vec![
        Field::new("id", DataType::Int32, true),
        Field::new("payload", variant_type, true),
    ]));

    let sdk = ZerobusSdk::builder()
        .endpoint(ENDPOINT)
        .unity_catalog_url(UC_URL)
        .build()?;

    let mut stream = sdk
        .stream_builder()
        .table(TABLE)
        .oauth(&client_id, &client_secret)
        .arrow(schema.clone())
        .ipc_compression(None)
        .build_arrow()
        .await?;

    let ids = Int32Array::from(vec![1, 2, 3]);
    let payload = variant_struct(&[
        r#"{"user":"alice","tags":[1,2,3]}"#,
        r#""just a string""#,
        r#"{"nested":{"a":true,"b":null,"c":3.14}}"#,
    ]);
    let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(ids) as ArrayRef, payload])?;

    let offset = stream.ingest_batch(batch).await?;
    stream.flush().await?;
    stream.close().await?;
    Ok(())
}

Este exemplo está em Rust. Para uso equivalente em outras linguagens, consulte o repository do SDK Zerobus.

Compressão IPC​

Por padrão, os payloads do Arrow IPC são enviados sem compressão. É possível, opcionalmente, compactá-los no tráfego utilizando um de dois codecs.

  • LZ4_FRAME: rápido, baixa sobrecarga de CPU, taxa de compressão modesta. Prefira isto quando o cliente estiver com restrições de CPU, mas ainda desejar reduzir os bytes na rede.
  • ZSTD: Maior taxa de compressão, mais CPU por lote. Recomenda-se habilitá-lo sempre que o cliente puder arcar com o custo adicional da CPU.

A compressão reduz bytes na rede, mas acarreta um custo de CPU no cliente. Cargas menores podem evitar gargalos de rede e reduzir os custos de rede.

No SDK do Python, defina o campo ipc_compression em ArrowStreamConfigurationOptions:

Python
from zerobus.sdk.shared.arrow import IPCCompression, ArrowStreamConfigurationOptions

options = ArrowStreamConfigurationOptions(ipc_compression=IPCCompression.ZSTD)

No SDK do Rust, configure-o no construtor. O enum CompressionType reside no crate arrow-ipc , portanto, adicione-o como uma dependência:

Bash
cargo add arrow-ipc
Rust
use arrow_ipc::CompressionType;

let stream = sdk
    .stream_builder()
    .table(TABLE_NAME)
    .oauth(CLIENT_ID, CLIENT_SECRET)
    .arrow(schema)
    .ipc_compression(Some(CompressionType::ZSTD))
    .build_arrow()
    .await?;

Melhores práticas​

Siga estas orientações para obter o melhor desempenho e confiabilidade da ingestão de dados do Arrow Flight.

  • Reutilizar uma transmissão por muitos lotes em vez de abrir uma nova transmissão por lotes. A criação de transmissões acarreta custos indiretos significativos que podem ser amortizados reutilizando uma transmissão em vários lotes.
  • Enviar várias linhas por lote. Comece com lotes do tamanho ideal para a aplicação, e não com uma linha por chamada. Enviar uma linha de cada vez funciona, mas anula a maior parte da vantagem de desempenho de usar o Arrow.
  • Chame flush() em pontos de verificação controlados. Isso proporciona um limite de durabilidade claro para um grupo de lotes, sem bloquear cada um individualmente.
  • Habilite a compressão IPC para melhorar o throughput. ZSTD é recomendado para a maioria das cargas de trabalho quando o cliente tem CPU disponível. Use LZ4_FRAME ou nenhuma compressão se o cliente estiver com a CPU limitada.
  • Use o Arrow Flight quando seu produtor já for colunar. Se seus dados de origem forem naturalmente orientados a linhas e pequenos, usar o Zerobus Ingest com JSON ou protobuf costuma ser mais simples. Consulte Use Zerobus Ingest.

Tratamento e recuperação de erros​

As transmissões do Arrow Flight usam as mesmas categorias de erro gRPC que o restante do Zerobus Ingest. Para obter informações sobre códigos de erro, orientações sobre como tentar novamente e a taxonomia completa de cliente versus servidor, consulte Tratamento de erros do Zerobus Ingest.

Ao configurar o SDK com recuperação automática (o default), ele se reconecta e reproduz lotes não reconhecidos de forma transparente em caso de falhas transitórias. Após o encerramento da transmissão, você poderá recuperar quaisquer lotes que o servidor tenha recebido, mas ainda não tenha confirmado. Isso se aplica tanto se a transmissão foi encerrada normalmente quanto devido a uma falha irrecuperável. No SDK do Python:

Python
# Retry unacked_batches against a freshly created stream
if stream.is_closed:
    unacked_batches = stream.get_unacked_batches()

No SDK Rust, chame stream.get_unacked_batches().await? para recuperar lotes não reconhecidos para nova tentativa.

Recursos adicionais​

  • Use Zerobus Ingest: Se você ainda não configurou o Zerobus Ingest, comece aqui para obter instruções sobre como encontrar a URL do seu workspace, criar a tabela Delta de destino e configurar um Service Principal. Estes passos são compartilhados entre todos os formatos de registro.
  • Cotas do Zerobus Ingest: revise as cotas default do Zerobus antes de implantar em produção. As mesmas características de throughput e latência se aplicam ao Arrow Flight, e escalam para atender a cargas de trabalho maiores.
  • Tratamento de erros do Zerobus Ingest: Consulte esta página para obter uma lista completa dos códigos de erro gRPC e o comportamento recomendado de repetição e recuperação para o seu cliente.
Nesta página