Características de desempenho dos pipelines do Kafka para o BigQuery

Esta página descreve as características de desempenho dos jobs de streaming do Dataflow que leem do Apache Kafka e gravam no BigQuery. Ela apresenta os resultados de testes de comparação para pipelines somente de mapa, que realizam transformações por mensagem sem rastrear o estado ou agrupar elementos no stream.

Muitas cargas de trabalho de integração de dados, incluindo ETL, validação de campo e mapeamento de esquema, se enquadram na categoria somente de mapa. Se o pipeline seguir esse padrão, você poderá usar essas comparações para avaliar seu job do Dataflow em relação a uma configuração de referência com bom desempenho.

Metodologia de teste

As comparações foram realizadas usando os seguintes recursos:

  • Um cluster do Serviço Gerenciado para Apache Kafka. As mensagens foram geradas usando o modelo do Gerador de dados de streaming.

    • Taxa de mensagens: aproximadamente 1.000.000 de mensagens por segundo
    • Carga de entrada: 1 GiB/s
    • Formato de mensagem: texto JSON gerado aleatoriamente com um esquema fixo
    • Tamanho da mensagem: aproximadamente 1 KiB por mensagem
    • Partições do Kafka: 1000

  • Uma tabela padrão do BigQuery.

  • Um pipeline de streaming do Dataflow que usou o modelo do Apache Kafka para BigQuery. Esse pipeline realiza a análise e o mapeamento de esquema mínimos necessários. Nenhuma função definida pelo usuário (UDF) personalizada foi usada.

Depois que o escalonamento horizontal se estabilizou e o pipeline atingiu o estado estável, os pipelines puderam ser executados por aproximadamente um dia. Depois disso, os resultados foram coletados e analisados.

Pipeline do Dataflow

Essa comparação usa um pipeline somente de mapa que realiza um mapeamento e uma conversão simples de mensagens JSON. O pipeline foi testado usando o modo "exatamente uma vez" e o modo "pelo menos uma vez". O processamento "pelo menos uma vez" oferece melhor capacidade de processamento. No entanto, ele só deve ser usado quando registros duplicados forem aceitáveis ou quando o coletor downstream processar a remoção de duplicação.

Configuração do job

A tabela a seguir mostra como os jobs do Dataflow foram configurados.

Configuração Valor
Tipo de máquina do worker e2-standard-2
vCPUs da máquina do worker 2
RAM da máquina do worker 8 GB
Persistent Disk da máquina do worker Disco permanente padrão (HDD), 30 GB
Número máximo de workers 120
Streaming Engine Sim
Escalonamento automático horizontal Sim
Modelo de faturamento Faturamento baseado em recursos
A API Storage Write (gRPC) está ativada? Sim
Streams da API Storage Write (gRPC) 400
Frequência de acionamento da API Storage Write (gRPC) 5 segundos
Formato de mensagem JSON
Modo de autenticação do Kafka

Application Default Credentials (ADC).

Para mais informações, consulte Tipos de autenticação para brokers do Kafka.

A API BigQuery Storage Write (gRPC) é recomendada para pipelines de streaming. Ao usar o modo "exatamente uma vez" com a API Storage Write (gRPC), é possível ajustar as seguintes configurações:

Para mais informações, consulte Gravar do Dataflow para o BigQuery.

Uma consideração especial também deve ser dada ao número de partições do Apache Kafka. Para garantir paralelismo de chave suficiente no estágio de leitura, o número de partições precisa ser pelo menos igual ao número total de vCPUs de worker. Para mais informações, consulte Ler do Apache Kafka para o Dataflow.

Resultados da comparação

Esta seção descreve os resultados dos testes de comparação.

Capacidade de processamento e uso de recursos

A tabela a seguir mostra os resultados do teste para capacidade de processamento e uso de recursos do pipeline.

Resultado Exatamente uma vez Pelo menos uma vez
Capacidade de processamento de entrada por worker Média: 15 MBps, n=3 Média: 18 MBps, n=3
Uso médio da CPU em todos os workers Média: 70%, n=3 Média: 75%, n=3
Número de nós de trabalho Média: 63, n=3 Média: 53, n=3
Unidades de computação do Streaming Engine por hora Média: 58, n=3 Média: 0, n=3

O algoritmo de escalonamento automático pode afetar o nível de uso da CPU de destino. Para alcançar um uso maior ou menor da CPU de destino, defina o intervalo de escalonamento automático ou a dica de uso do worker. Metas de uso mais altas podem levar a custos mais baixos, mas também a uma latência de cauda pior, especialmente para cargas variáveis.

Latência

A tabela a seguir mostra os resultados da comparação da latência do pipeline para o modo "exatamente uma vez", excluindo o estágio de entrada.

Latência total de ponta a ponta do estágio, excluindo o estágio de entrada Exatamente uma vez
P50 Média: 1.200 ms, n=3
P95 Média: 3.000 ms, n=3
P99 Média: 5.400 ms, n=3

Os testes mediram a latência de ponta a ponta por estágio (a job/streaming_engine/stage_end_to_end_latencies métrica) em três execuções de teste de longa duração. Essa métrica mede quanto tempo o Streaming Engine gasta em cada estágio do pipeline. Ela abrange todas as etapas internas do pipeline, como:

  • Embaralhamento e enfileiramento de mensagens para processamento
  • O tempo de processamento real; por exemplo, converter mensagens em objetos de linha
  • Gravação de estado persistente, bem como tempo gasto no enfileiramento para gravar o estado persistente

Devido a uma limitação da métrica, a latência do estágio de entrada não é informada. Portanto, ela não está incluída no total.

As comparações mostradas aqui representam uma linha de base. A latência é altamente sensível à complexidade do pipeline. UDFs personalizados, transformações adicionais e lógica de janelas complexa podem aumentar a latência.

Estimar custos

É possível estimar o custo de referência do seu próprio pipeline comparável com o faturamento baseado em recursos usando a Google Cloud calculadora de preços, da seguinte maneira:

  1. Abra a calculadora de preços.
  2. Clique em Adicionar à estimativa.
  3. Selecione Dataflow.
  4. Em Tipo de serviço, selecione "Dataflow Classic".
  5. Selecione Configurações avançadas para mostrar o conjunto completo de opções.
  6. Escolha o local em que o job é executado.
  7. Em Tipo de serviço, selecione "Streaming".
  8. Selecione Ativar o Streaming Engine.
  9. Insira informações sobre as horas de execução do job, nós de trabalho, máquinas de worker e armazenamento em Persistent Disk.
  10. Insira o número estimado de unidades de computação do Streaming Engine.

O uso de recursos e o custo são escalonados de forma aproximadamente linear com a capacidade de processamento de entrada, embora, para jobs pequenos com apenas alguns workers, o custo total seja dominado por custos fixos. Como ponto de partida, é possível extrapolar o número de nós de trabalho e o consumo de recursos dos resultados da comparação.

Por exemplo, suponha que você execute um pipeline somente de mapa no modo "exatamente uma vez", com uma taxa de dados de entrada de 100 MiB/s. Com base nos resultados da comparação de um pipeline de 1 GiB/s, é possível estimar os requisitos de recursos da seguinte maneira:

  • Fator de escalonamento: (100 MiB/s) / (1 GiB/s) = 0,1
  • Nós de trabalho projetados: 63 workers × 0,1 = 6,3 workers
  • Número projetado de unidades de computação do Streaming Engine por hora: 58 × 0,1 = 5,8 unidades por hora

Esse valor só deve ser usado como uma estimativa inicial. A capacidade de processamento e o custo reais podem variar significativamente, com base em fatores como tipo de máquina, distribuição de tamanho de mensagem, código do usuário, tipo de agregação, paralelismo de chave e tamanho da janela. Para mais informações, consulte Práticas recomendadas para otimização de custos do Dataflow.

Executar um pipeline de teste

Esta seção mostra os gcloud dataflow flex-template run comandos que foram usados para executar o pipeline somente de mapa.

Modo "exatamente uma vez" 

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Kafka_to_BigQuery_Flex \
  --enable-streaming-engine \
  --parameters \
readBootstrapServerAndTopic="KAFKA_BOOTSTRAP_ADDRESS;KAFKA_TOPIC",\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
writeMode=SINGLE_TABLE_NAME,\
outputTableSpec="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME",\
useBigQueryDLQ=true,\
outputDeadletterTable="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME_dlq",\
numStorageWriteApiStreams=400

Modo "pelo menos uma vez" 

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Kafka_to_BigQuery_Flex \
  --enable-streaming-engine \
  --additional-experiments=streaming_mode_at_least_once \
  --parameters \
readBootstrapServerAndTopic="KAFKA_BOOTSTRAP_ADDRESS;KAFKA_TOPIC",\
kafkaReadAuthenticationMode=APPLICATION_DEFAULT_CREDENTIALS,\
messageFormat=JSON,\
writeMode=SINGLE_TABLE_NAME,\
outputTableSpec="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME",\
useBigQueryDLQ=true,\
outputDeadletterTable="PROJECT_ID:BQ_DATASET.BQ_TABLE_NAME_dlq",\
numStorageWriteApiStreams=400,\
useStorageWriteApiAtLeastOnce=true

Substitua:

  • JOB_NAME: o nome do job do Dataflow
  • PROJECT_ID: o ID do projeto
  • KAFKA_BOOTSTRAP_ADDRESS: o endereço de inicialização do cluster do Apache Kafka
  • KAFKA_TOPIC: o nome do tópico do Kafka
  • BQ_DATASET: o nome do conjunto de dados do BigQuery
  • BQ_TABLE_NAME: o nome da tabela do BigQuery

Gerar dados de teste

Para gerar dados de teste, use o comando a seguir para executar o modelo do Gerador de dados de streaming:

gcloud dataflow flex-template run JOB_NAME \
  --project=PROJECT_ID \
  --template-file-gcs-location=gs://dataflow-templates-us-central1/latest/flex/Streaming_Data_Generator \
  --max-workers=140 \
  --parameters \
schemaLocation=SCHEMA_LOCATION,\
qps=1000000,\
sinkType=KAFKA,\
bootstrapServer=KAFKA_BOOTSTRAP_ADDRESS,\
kafkaTopic=KAFKA_TOPIC,\
outputType=JSON

Substitua:

  • JOB_NAME: o nome do job do Dataflow
  • PROJECT_ID: o ID do projeto
  • SCHEMA_LOCATION: o caminho para um arquivo de esquema no Cloud Storage
  • KAFKA_BOOTSTRAP_ADDRESS: o endereço de inicialização do cluster do Apache Kafka
  • KAFKA_TOPIC: o nome do tópico do Kafka

O modelo do Gerador de dados de streaming usa um arquivo do Gerador de dados JSON para definir o esquema de mensagens. Os testes de comparação usaram um esquema de mensagens semelhante ao seguinte:

{
  "logStreamId": "{{integer(1000001,2000000)}}",
  "message": "{{alphaNumeric(962)}}"
}

Próximas etapas