Realtime Compute for Apache Flink:Log Service (SLS)

A tabela a seguir lista as capacidades do conector SLS.

O conector de source SLS lê diretamente os campos de atributos das mensagens. A tabela abaixo detalha os campos suportados.

Sintaxe

CREATE TABLE sls_table(
  a INT,
  b INT,
  c VARCHAR
) WITH (
  'connector' = 'sls',
  'endPoint' = '<yourEndPoint>',
  'project' = '<yourProjectName>',
  'logStore' = '<yourLogStoreName>',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}'
);

Opções With

• Gerais

  Parâmetro	Descrição	Tipo	Obrigatório	Padrão	Observações
  connector	Conector a ser usado.	String	Sim	Nenhum	Defina como sls.
  endPoint	Endpoint do Log Service (SLS).	String	Sim	Nenhum	Especifique o endereço de acesso VPC do Log Service (SLS). Para mais informações, consulte Service endpoints. Nota Por padrão, o Realtime Compute for Apache Flink não acessa a internet. Para habilitar o acesso à internet a partir da sua Virtual Private Cloud (VPC), use um NAT Gateway. Para mais detalhes, veja How do I access the Internet?. Recomendamos não acessar o SLS pela internet. Caso seja necessário, use HTTPS e ative a aceleração de transferência. Consulte Manage transfer acceleration para saber mais.
  project	Nome do projeto SLS.	String	Sim	Nenhum	N/A
  logStore	Nome do Logstore ou MetricStore.	String	Sim	Nenhum	Os dados em um Logstore são consumidos da mesma forma que os dados em um MetricStore.
  accessId	AccessKey ID da sua conta Alibaba Cloud.	String	Sim	Nenhum	Para mais informações, consulte Obtain an AccessKey pair. Importante Para evitar a exposição do seu par AccessKey, recomendamos usar variáveis para especificar o AccessKey ID e o AccessKey secret. Veja Project variables para detalhes.
  accessKey	AccessKey secret da sua conta Alibaba Cloud.	String	Sim	Nenhum

• Específicas da source

  Parâmetro	Descrição	Tipo	Obrigatório	Padrão	Observações
  enableNewSource	Define se deve ser usada a nova fonte de dados que implementa a interface FLIP-27.	Boolean	Não	false	A nova fonte adapta-se automaticamente a mudanças nos shards e distribui-os da forma mais equilibrada possível entre todas as subtarefas de source. Importante Esta opção é suportada apenas no VVR 8.0.9 e posteriores. O valor padrão é true para VVR 11,1 e superiores. Se você alterar o valor desta opção, não será possível restaurar jobs a partir de um estado salvo. Para retomar o consumo de um offset histórico, inicie primeiro o job com a opção consumerGroup para registrar o progresso de consumo no grupo de consumidores do SLS. Em seguida, defina a opção consumeFromCheckpoint como true e reinicie o job sem estado. Caso um Logstore possua shards somente leitura, algumas subtarefas podem continuar solicitando dados de outros shards após concluírem os seus próprios. Isso pode gerar desequilíbrio na carga de trabalho e afetar o desempenho. Para mitigar esse problema, ajuste o paralelismo, otimize a estratégia de agendamento ou mescle shards pequenos para reduzir a quantidade total e simplificar a alocação de tarefas.
  shardDiscoveryIntervalMs	Intervalo para descoberta dinâmica de shards.	Long	Não	60000	Defina esta opção com um valor negativo para desativar a descoberta dinâmica. Unidade: milissegundos. Nota O valor deve ser maior ou igual a 60.000 milissegundos (1 minuto). Esta opção só tem efeito quando enableNewSource está definido como true. Suportado apenas no VVR 8.0.9 e versões posteriores.
  startupMode	Modo de inicialização da tabela de source.	String	Não	timestamp	timestamp (padrão): Consome logs a partir do horário inicial especificado. latest: Inicia o consumo de logs pelo offset mais recente. earliest: Começa o consumo de logs pelo offset mais antigo. consumer_group: Inicia o consumo a partir do offset registrado pelo grupo de consumidores. Se o grupo não tiver um offset registrado para determinado shard, o consumo começa pelo offset mais antigo. Importante Para versões do VVR anteriores à 11,1, o valor consumer_group não é suportado. É necessário definir consumeFromCheckpoint como true. Nesse caso, o consumo de logs inicia no offset registrado pelo grupo de consumidores especificado, e a configuração do modo de inicialização não terá efeito.
  startTime	Horário inicial para consumo de logs.	String	Não	Horário atual	O formato é yyyy-MM-dd hh:mm:ss. Tem efeito apenas quando startupMode está definido como timestamp. Nota As opções startTime e stopTime baseiam-se no atributo __receive_time__ do SLS, e não no atributo __timestamp__.
  stopTime	Horário final para consumo de logs.	String	Não	Nenhum	O formato é yyyy-MM-dd hh:mm:ss. Nota Esta opção serve apenas para consumir logs históricos e deve ser configurada com uma data passada. Se definida para um horário futuro, o consumo poderá parar prematuramente caso nenhum novo log seja gravado, causando interrupção no fluxo de dados sem mensagens de erro. Para que o job Flink encerre após o consumo de todos os logs, também é necessário definir exitAfterFinish como true.
  consumerGroup	Nome do grupo de consumidores.	String	Não	Nenhum	Um grupo de consumidores registra o progresso do consumo. Você pode definir um nome personalizado sem formato fixo. Nota Jobs Flink diferentes devem usar grupos de consumidores distintos. Se múltiplos jobs utilizarem o mesmo grupo, eles não se coordenarão e cada um consumirá todos os dados. Isso ocorre porque o Flink não usa o grupo de consumidores do SLS para atribuição de partições ao consumir dados. Consequentemente, cada consumidor processa mensagens independentemente, mesmo compartilhando o mesmo grupo.
  consumeFromCheckpoint	Define se o consumo deve iniciar a partir de um checkpoint do grupo de consumidores.	String	Não	false	true: Também é necessário especificar um grupo de consumidores. O programa Flink inicia o consumo dos logs a partir do checkpoint salvo nesse grupo. Caso não exista um checkpoint correspondente, o consumo começa pelo valor configurado em startTime. false (valor padrão): Não inicia o consumo de logs a partir do checkpoint salvo para o grupo de consumidores especificado. Importante Este parâmetro não é mais suportado no VVR 11,1 e versões posteriores. Para essas versões, defina a opção startupMode como consumer_group.
  maxRetries	Número de tentativas após falha na leitura do SLS.	String	Não	3	N/A
  batchGetSize	Quantidade de grupos de logs lidos por requisição.	String	Não	100	A configuração batchGetSize não pode exceder 1000, caso contrário um erro será reportado.
  exitAfterFinish	Define se o job Flink encerra após todo o dado ser consumido.	String	Não	false	true: O programa Flink encerra após o consumo completo dos dados. false (padrão): O programa Flink não encerra após a conclusão do consumo de dados.
  query Importante Esta opção foi descontinuada no VVR 11.3, mas versões posteriores permanecem compatíveis.	Instrução de consulta para pré-processamento de dados antes do consumo.	String	Não	Nenhum	Use esta opção para filtrar dados do SLS antes que o Flink os consuma. Isso reduz custos e aumenta a velocidade de processamento. Por exemplo, 'query' = '*| where request_method = ''GET''' indica que, antes de ler dados do SLS, o Flink selecionará apenas registros onde o campo request_method tenha o valor 'GET'. Nota Esta opção utiliza a linguagem SPL do Log Service (SLS). Para mais informações, consulte SPL syntax. Importante Suportado apenas no VVR 8.0.1 e versões posteriores. Este recurso gera cobranças do Log Service (SLS). Consulte Billing of Log Service para detalhes.
  processor	Nome do processador SLS para pré-processamento de dados. Se tanto query quanto processor forem especificados, query terá precedência e processor será ignorado.	String	Não	Nenhum	Esta opção filtra dados do SLS antes do consumo pelo Flink, reduzindo custos e melhorando a velocidade de processamento. Recomendamos o uso de processor em vez de query. Por exemplo, 'processor' = 'test-filter-processor' indica que o processador SLS filtrará os dados antes que o Flink faça a leitura. Nota Esta opção utiliza a linguagem SPL do Log Service (SLS). Para mais informações, consulte SPL syntax. Para saber como criar ou atualizar um processador SLS, veja Manage processors. Importante Suportado apenas no VVR 11.3 e versões posteriores. Este recurso gera cobranças do Log Service (SLS). Consulte Billing of Log Service para detalhes.
  preserveRawBytes	Define se os bytes brutos transportados pelo SLS devem ser buscados e preservados diretamente quando enableNewSource estiver definido como true.	Boolean	Não	false	Quando esta opção está ativada, campos BINARY/VARBINARY leem o byte[] bruto diretamente, e campos CHAR/VARCHAR constroem strings a partir dos bytes brutos. Outros campos continuam usando a lógica de conversão padrão. O comportamento é consistente com o da fonte antiga. Nota Suportado apenas no VVR 11.8 e versões posteriores. Esta opção só tem efeito quando enableNewSource está definido como true.

• Específicas do destino

  Parâmetro	Descrição	Tipo	Obrigatório	Padrão	Observações
  topicField	Especifica um campo cujo valor sobrescreve o atributo __topic__, indicando o tópico do log.	String	Não	Nenhum	O valor desta opção deve ser um campo existente na tabela.
  timeField	Especifica um campo cujo valor sobrescreve o atributo __timestamp__, indicando o horário de gravação do log.	String	Não	Horário atual	O valor desta opção deve ser um campo INT existente na tabela. Se não especificado, o horário atual será usado.
  sourceField	Especifica um campo cujo valor sobrescreve o atributo __source__, indicando a source do log, como o endereço IP da máquina que o gerou.	String	Não	Nenhum	O valor desta opção deve ser um campo existente na tabela.
  partitionField	Define um campo para particionamento. Um hash do valor deste campo determina qual shard receberá os dados, garantindo que registros com o mesmo hash vão para o mesmo shard.	String	Não	Nenhum	Se esta opção não for especificada, cada registro será gravado aleatoriamente em um shard disponível.
  buckets	Se partitionField for especificado, esta opção define o número de buckets para mapeamento de valores hash.	String	Não	64	O valor deve ser uma potência de 2 no intervalo [1, 256]. O número de buckets deve ser maior ou igual ao número de shards. Caso contrário, alguns shards podem não receber dados.
  flushIntervalMs	Intervalo de gravação de dados.	String	Não	2000	Unidade: milissegundos.
  writeNullProperties	Define se valores nulos devem ser gravados como strings vazias no SLS.	Boolean	Não	true	true (valor padrão): Grava valores nulos no log como strings vazias. false: Campos avaliados como nulo não são gravados no log. Nota Suportado apenas no VVR 8.0.6 e versões posteriores.

Mapeamento de tipos

Limitações

Este recurso é suportado apenas pelo Realtime Compute for Apache Flink versão 11,1 e posteriores.

Sintaxe

source:
   type: sls
   name: SLS Source
   endpoint: <endpoint>
   project: <project>
   logstore: <logstore>
   accessId: <accessId>
   accessKey: <accessKey>

Parâmetros

Reutilizar um catálogo existente

A partir da versão 11,5 do Realtime Compute for Apache Flink, é possível referenciar um Catálogo SLS integrado criado na página Data Management em jobs de ingestão de dados Flink CDC. Isso elimina a necessidade de especificar manualmente as propriedades de conexão.

source:
  type: sls
  using.built-in-catalog: sls_catalog

Atualmente, os jobs de ingestão de dados podem reutilizar automaticamente os seguintes parâmetros do Catálogo SLS integrado:

• endpoint

• project

• accessId

• accessKey

Para substituir esses parâmetros reutilizados automaticamente, defina-os explicitamente na configuração yaml. Os parâmetros definidos no arquivo yaml têm precedência.

Mapeamento de tipos de dados

Se fixed-types não estiver configurado, aplica-se o seguinte mapeamento de tipos de dados:

Se fixed-types estiver configurado, o sistema analisa os dados utilizando os tipos especificados.

Inferência e evolução de schema

• Pré-consumo de dados de shard e inicialização de schema

  O conector SLS mantém o schema do Logstore que está lendo. Antes de ler os dados, o conector pré-consome até maxPreFetchLogGroups grupos de logs de cada shard. Ele analisa o schema de cada entrada de log e os mescla para inicializar o schema da tabela. Um evento de criação de tabela é então gerado com base nesse schema inicial antes do início do consumo de dados.

  Nota

  Para cada shard, o conector tenta consumir dados começando de uma hora antes do horário atual para analisar o schema do log.

• Informações de chave primária

  Logs do Log Service (SLS) não contêm informações de chave primária. Adicione manualmente uma chave primária à tabela usando regras de transformação:

  transform:
    - source-table: <project>.<logstore>
      projection: *
      primary-keys: key1, key2

• Inferência de schema e alterações de schema

  Após a inicialização do schema, se schema.inference.strategy estiver definido como static, o conector SLS analisa cada entrada de log com base no schema inicial e não gera eventos de alteração de schema. Se schema.inference.strategy estiver definido como continuous, o conector analisa cada entrada, infere as colunas físicas e as compara com o schema atual. Se o schema inferido for inconsistente com o atual, os schemas são mesclados conforme as seguintes regras:

  • Se as colunas físicas inferidas contiverem campos que não estão no schema atual, o conector adiciona esses campos ao schema e gera um evento para adicionar colunas anuláveis.

  • Se as colunas físicas inferidas não tiverem campos que existem no schema atual, o conector retém esses campos, preenche seus dados com NULL e não gera evento de exclusão de coluna.

  O conector SLS infere o tipo de dados de todos os campos em cada entrada de log como String. Atualmente, apenas a adição de novas colunas é suportada. O conector anexa novas colunas ao final do schema e as define como anuláveis.

source:
  type: sls
  name: SLS Source
  endpoint: ${endpoint}
  project: test_project
  logstore: test_log
  accessId: ${accessId}
  accessKey: ${accessKey}
   
# Add a primary key to the table.
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id
    
# Route all data from test_project.test_log to the test_database.inventory table.
route:
  - source-table: test_project.test_log
    sink-table: test_database.inventory

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true