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 a seguir lista 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	O conector a ser usado.	String	Sim	Nenhum	Defina como sls.
  endPoint	O endpoint do Log Service (SLS).	String	Sim	Nenhum	Especifique o endereço de acesso vpc do Log Service (SLS). Para mais informações, consulte Endpoints de serviço. 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 informações, consulte Como acesso a Internet?. Recomendamos não acessar o SLS pela internet. Caso seja necessário, use HTTPS e ative a aceleração de transferência. Para mais informações, consulte Gerenciar aceleração de transferência.
  project	O nome do projeto SLS.	String	Sim	Nenhum	N/A
  logStore	O 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	O AccessKey ID da sua conta Alibaba Cloud.	String	Sim	Nenhum	Para mais informações, consulte Obter um par de AccessKey. Importante Para evitar a exposição do seu par de AccessKey, recomendamos o uso de variáveis para especificar o AccessKey ID e o AccessKey secret. Para mais informações, consulte Variáveis de projeto.
  accessKey	O AccessKey secret da sua conta Alibaba Cloud.	String	Sim	Nenhum

• Específicas da origem

  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 alterações nos shards e distribui os shards da forma mais uniforme possível entre todas as subtarefas de origem. Importante Esta opção é suportada apenas no VVR 8.0.9 e posteriores. O valor padrão é true para VVR 11,1 e posteriores. 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 a partir 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. Se um Logstore possuir shards somente leitura, algumas subtarefas podem continuar solicitando dados de outros shards após concluírem os seus próprios. Isso pode causar desequilíbrio na carga de trabalho, afetando o desempenho. Para mitigar esse problema, ajuste o paralelismo, otimize a estratégia de agendamento ou mescle shards pequenos para reduzir o número total de shards e simplificar a alocação de tarefas.
  shardDiscoveryIntervalMs	O 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. Esta opção é suportada apenas no VVR 8.0.9 e posteriores.
  startupMode	O modo de inicialização da tabela de origem.	String	Não	timestamp	timestamp (padrão): Consome logs a partir da hora de início especificada. latest: Inicia o consumo de logs a partir do offset mais recente. earliest: Inicia o consumo de logs a partir do offset mais antigo. consumer_group: Inicia o consumo de logs a partir do offset registrado pelo grupo de consumidores. Se o grupo de consumidores não tiver registrado um offset de consumo para um shard, o consumo começa a partir do 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 a partir do offset registrado pelo grupo de consumidores especificado, e a configuração do modo de inicialização não terá efeito.
  startTime	A hora de início para o consumo de logs.	String	Não	Hora atual	O formato é yyyy-MM-dd hh:mm:ss. Isso só tem efeito quando startupMode está definido como timestamp. Nota As opções startTime e stopTime baseiam-se no atributo __receive_time__ no SLS, e não no atributo __timestamp__.
  stopTime	A hora de término para o 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 definida com uma hora no passado. Se definida para uma hora futura, o consumo poderá parar prematuramente caso nenhum novo log seja gravado, resultando em interrupção do fluxo de dados sem mensagens de erro. Se desejar que o job Flink encerre após o consumo de todos os logs, também defina exitAfterFinish como true.
  consumerGroup	O nome do grupo de consumidores.	String	Não	Nenhum	Um grupo de consumidores registra o progresso de consumo. É possível especificar um nome personalizado sem formato fixo. Nota Jobs Flink diferentes devem usar grupos de consumidores distintos. Se múltiplos jobs Flink usarem o mesmo grupo de consumidores, eles não se coordenarão e cada job 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 do SLS. Como resultado, cada consumidor processa mensagens independentemente, mesmo compartilhando o mesmo grupo de consumidores.
  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 de logs a partir do checkpoint salvo no grupo de consumidores. Se o grupo não possuir um checkpoint correspondente, o consumo começa a partir do 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 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 em uma única solicitaçã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 o consumo de todos os dados.	String	Não	false	true: O programa Flink encerra após o consumo de todos os 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 seleciona apenas os registros onde o valor do campo request_method é 'GET'. Nota Esta opção usa a linguagem SPL do Log Service (SLS). Para mais informações, consulte Sintaxe SPL. Importante Esta opção é suportada apenas no VVR 8.0.1 e posteriores. Este recurso gera cobranças do Log Service (SLS). Para mais informações, consulte Faturamento do Log Service.
  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, o que reduz custos e melhora a velocidade de processamento. Recomendamos o uso de processor em vez de query. Por exemplo, 'processor' = 'test-filter-processor' indica que o processador SLS filtra os dados antes que o Flink leia do SLS. Nota Esta opção usa a linguagem SPL do Log Service (SLS). Para mais informações, consulte Sintaxe SPL. Para saber como criar ou atualizar um processador SLS, consulte Gerenciar processadores. Importante Esta opção é suportada apenas no VVR 11.3 e posteriores. Este recurso gera cobranças do Log Service (SLS). Para mais informações, consulte Faturamento do Log Service.

• 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__, que indica 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__, que indica a hora de gravação do log.	String	Não	Hora atual	O valor desta opção deve ser um campo INT existente na tabela. Se esta opção não for especificada, a hora atual será usada.
  sourceField	Especifica um campo cujo valor sobrescreve o atributo __source__, que indica a origem do log, como o endereço IP da máquina que gerou o log.	String	Não	Nenhum	O valor desta opção deve ser um campo existente na tabela.
  partitionField	Especifica um campo para particionamento. Um hash do valor deste campo determina qual shard recebe os dados, garantindo que registros com o mesmo hash sejam enviados ao 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 dos valores de 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	O 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 Esta opção é suportada apenas no VVR 8.0.6 e posteriores.

Mapeamentos de tipo

Limitações

Este recurso é suportado apenas pelas versões 11,1 e posteriores do Realtime Compute for Apache Flink.

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 evita 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 usando 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 dados do Logstore, 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 neste 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 da hora atual para analisar o schema do log.

• Informações de chave primária

  Os 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 de log, infere as colunas físicas e as compara com o schema atual. Se o schema inferido for inconsistente com o schema atual, os schemas são mesclados de acordo com 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 possuírem campos existentes no schema atual, o conector retém esses campos, preenche seus dados com NULL e não gera um 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