Realtime Compute for Apache Flink:Log Service (SLS)

A tabela a seguir lista os recursos do conector SLS.

O conector 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	Selecione 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, acesse Service endpoints. Nota Por padrão, o Realtime Compute for Apache Flink não pode acesse a internet. Ative o acesso à internet a partir da sua Virtual Private cloud (VPC), use um NAT Gateway. Para mais informações, acesse How do I access the Internet?. Recomendamos que você não acesse o SLS pela internet. Se for necessário, use HTTPS e ative a aceleração de transferência. Para mais informações, acesse Manage transfer acceleration.
  project	O nome do projeto SLS.	String	Sim	Nenhum	N/D
  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, acesse Obtain an accesskey pair. Importante Para evitar que seu par de accesskey seja exposto, recomendamos o uso de variáveis e especifique o accesskey ID e o accesskey secret. Para mais informações, acesse Project variables.
  accessKey	O accesskey secret da sua conta Alibaba cloud.	String	Sim	Nenhum
  akless.assume-role.role-name	O nome da função RAM usada para acesse o SLS.	String	Não	Nenhum	Se accessId e accessKey não forem especificados, você deve especifique este parâmetro para acesse os recursos do SLS usando uma função RAM. A função RAM deve ser referenciada previamente em Access Security/RAM Roles. Nota Este parâmetro é compatível apenas com o VVR 11.9.0 e versões posteriores.

• Específicos do source

  Parâmetro	Descrição	Tipo	Obrigatório	Padrão	Observações
  enableNewSource	Especifica se deve usar o novo source de dados que implementa a API FLIP-27.	Boolean	Não	false	O novo source pode se adaptar automaticamente às alterações de shard e distribuir os shards o mais uniformemente possível entre todas as subtarefas de source. Importante Esta opção é compatível apenas com o VVR 8.0.9 e versões posteriores. O valor padrão é true para o VVR 11.1 e versões posteriores. Se você alterar o valor desta opção, não poderá restaurar jobs a partir de um estado salve. Para retomar o consumo a partir de um offset histórico, primeiro inicie o job com a opção consumerGroup para registrar o progresso do 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 tiver shards somente leitura, algumas subtarefas podem continuar solicitando dados de outros shards após terminarem os seus. Isso pode causar uma carga de trabalho desequilibrada, 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 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ó entra em vigor quando enableNewSource é defina como true. Esta opção é compatível apenas com o VVR 8.0.9 e versões posteriores.
  startupMode	O modo de inicialização para a tabela de source.	String	Não	timestamp	timestamp (padrão): Consome logs a partir do horário de início especificado. 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. Você deve defina consumeFromCheckpoint como true. Nesse caso, o consumo de logs começa a partir do offset registrado pelo grupo de consumidores especificado, e a configuração do modo de inicialização não terá efeito.
  startTime	O horário de início para o consumo de logs.	String	Não	Horário atual	O formato é yyyy-MM-dd hh:mm:ss. Isso só entra em vigor quando startupMode é defina como timestamp. Nota As opções startTime e stopTime são baseadas no atributo __receive_time__ no SLS, e não no atributo __timestamp__.
  stopTime	O horário de término para o consumo de logs.	String	Não	Nenhum	O formato é yyyy-MM-dd hh:mm:ss. Nota Esta opção é usada apenas para consumir logs históricos e deve ser defina com um horário no passado. Se você a defina com um horário futuro, o consumo poderá parar prematuramente se nenhum novo log for gravado, resultando em uma interrupção no fluxo de dados sem mensagens de erro. Se quiser que o job Flink seja encerrado após todos os logs serem consumidos, você também deve defina exitAfterFinish como true.
  consumerGroup	O nome do grupo de consumidores.	String	Não	Nenhum	Um grupo de consumidores registra o progresso do consumo. Você pode especifique um nome personalizado sem um formato fixo. Nota Jobs Flink diferentes devem usar grupos de consumidores diferentes. Se vários jobs Flink usarem o mesmo grupo de consumidores, não haverá coordenação entre eles 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 consome mensagens de forma independente, mesmo que compartilhem o mesmo grupo de consumidores.
  consumeFromCheckpoint	Especifica se deve consumir a partir de um checkpoint do grupo de consumidores.	String	Não	false	true: Você também deve especifique um grupo de consumidores. O programa Flink começa a consumir logs a partir do checkpoint salve no grupo de consumidores. Se o grupo de consumidores não tiver um checkpoint correspondente, o consumo começa a partir do valor de configuração startTime. false (valor padrão): Não inicia o consumo de logs a partir do checkpoint salve 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, você deve defina a opção startupMode como consumer_group.
  maxRetries	O número de tentativas após uma falha ao ler do SLS.	String	Não	3	N/D
  batchGetSize	O número de grupos de logs a serem lidos em uma única solicitação.	String	Não	100	O valor de batchGetSize não pode exceder 1000. Caso contrário, um erro será relatado.
  exitAfterFinish	Especifica se o job Flink é encerrado após todos os dados serem consumidos.	String	Não	false	true: O programa Flink é encerrado após todos os dados serem consumidos. false (padrão): O programa Flink não é encerrado após a conclusão do consumo de dados.
  query Importante Esta opção foi descontinuada no VVR 11.3, mas as versões posteriores permanecem compatíveis.	A instrução de consulta para pré-processar 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 melhora a velocidade de processamento. Por exemplo, 'query' = '*| where request_method = ''GET''' indica que, antes de o Flink ler dados do SLS, ele primeiro corresponde aos dados onde o valor do campo request_method é 'GET'. Nota Esta opção usa a linguagem SPL do Log Service (SLS). Para mais informações, acesse SPL syntax. Importante Esta opção é compatível apenas com o VVR 8.0.1 e versões posteriores. Este recurso incorre em taxas do Log Service (SLS). Para mais informações, acesse Billing of Log Service.
  processor	O nome do processador SLS para pré-processamento de dados. Se query e processor forem especificados, query terá precedência e processor será ignorado.	String	Não	Nenhum	Esta opção filtra dados do SLS antes que o Flink os consuma, 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 de o Flink lê-los do SLS. Nota Esta opção usa a linguagem SPL do Log Service (SLS). Para mais informações, acesse SPL syntax. Para obter informações sobre como crie ou atualize um processador SLS, acesse Manage processors. Importante Esta opção é compatível apenas com o VVR 11.3 e versões posteriores. Este recurso incorre em taxas do Log Service (SLS). Para mais informações, acesse Billing of Log Service.
  preserveRawBytes	Especifica se deve buscar e preservar diretamente os bytes brutos transportados pelo SLS quando enableNewSource é defina como true.	Boolean	Não	false	Quando esta opção está ativada, os campos BINARY/VARBINARY leem o byte[] bruto diretamente, e os campos CHAR/VARCHAR constroem dados de string a partir dos bytes brutos. Outros campos ainda usam a lógica de conversão padrão. O comportamento é consistente com o do source antigo. Nota Esta opção é compatível apenas com o VVR 11.8 e versões posteriores. Esta opção só entra em vigor quando enableNewSource é defina como true.

• Específicos 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	Horário 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 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 defina o número de buckets para mapear 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	Especifica se deve gravar valores nulos como strings vazias no SLS.	Boolean	Não	true	true (valor padrão): Grava valores nulos no log como strings vazias. false: Os campos avaliados como nulos não são gravados no log. Nota Esta opção é compatível apenas com o VVR 8.0.6 e versões posteriores.

Mapeamentos de tipo

Limitações

Este recurso é compatível apenas com as 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, você pode 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 especifique 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 defina no arquivo yaml têm precedência.

Mapeamento de tipo de dados

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

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 do qual 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 crie tabela é então gerado com base neste schema inicial antes que o consumo de dados comece.

  Nota

  Para cada shard, o conector tenta consumir dados a partir de uma hora antes do horário 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 o schema ser inicializado, se schema.inference.strategy for defina 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 for defina 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 contiverem campos que existem no schema atual, o conector retém esses campos, preenche seus dados com NULL e não gera um evento de exclua coluna.

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