Simple Log Service:Consumir dados de log com o Realtime Compute for Apache Flink

O Simple Log Service armazena dados de log em tempo real e o Realtime Compute for Apache Flink os lê no modo streaming. Exemplo de entrada de log:

__source__:  11.85.*.199
__tag__:__receive_time__:  1562125591
__topic__:  test-topic
request_method:  GET
status:  200

Código de exemplo

O exemplo de rascunho SQL a seguir consome dados do Simple Log Service no Realtime Compute for Apache Flink.

CREATE TEMPORARY TABLE sls_input(
  request_method STRING,
  status BIGINT,
  `__topic__` STRING METADATA VIRTUAL,
  `__source__` STRING METADATA VIRTUAL,
  `__timestamp__` BIGINT METADATA VIRTUAL,
   __tag__ MAP<VARCHAR, VARCHAR> METADATA VIRTUAL,
  proctime as PROCTIME()
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'starttime' = '2023-08-30 00:00:00',
  'project' ='sls-test',
  'logstore' ='sls-input'
);

CREATE TEMPORARY TABLE sls_sink(
  request_method STRING,
  status BIGINT,
  `__topic__` STRING,
  `__source__` STRING,
  `__timestamp__` BIGINT ,
  receive_time BIGINT
) WITH (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou-intranet.log.aliyuncs.com',
  'accessId' = '${ak_id}',
  'accessKey' = '${ak_secret}',
  'project' ='sls-test',
  'logstore' ='sls-output'
);

INSERT INTO sls_sink
SELECT 
  request_method,
  status,
  `__topic__` ,
  `__source__` ,
  `__timestamp__` ,
  cast(__tag__['__receive_time__'] as bigint) as receive_time
FROM sls_input; 

Opções

• 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 da vpc do Log Service (SLS). Para mais informações, consulte Service endpoints. Nota Por padrão, o Realtime Compute for Apache Flink não pode acessar a internet. Para ativar o acesso à internet a partir da sua Virtual Private cloud (vpc), use um nat gateway. Para mais informações, consulte 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, consulte Manage transfer acceleration.
  project	O nome do projeto do 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, consulte Obtain an AccessKey pair. Importante Para evitar que seu par de AccessKey seja exposto, recomendamos o uso de variáveis para especificar o AccessKey ID e o AccessKey secret. Para mais informações, consulte 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 acessar o SLS.	String	Não	Nenhum	Se accessId e accessKey não forem especificados, especifique este parâmetro para acessar os recursos do SLS com uma função RAM. Referencie a função RAM antecipadamente em Access Security/RAM Roles. Nota Este parâmetro tem suporte apenas no VVR 11.9.0 e versões posteriores.

• Específicas da source

  Parâmetro	Descrição	Tipo	Obrigatório	Padrão	Observações
  enableNewSource	Especifica se o job deve usar a nova source de dados que implementa a interface FLIP-27.	Boolean	Não	false	A nova source adapta-se automaticamente às alterações de shards e distribui os shards da forma mais uniforme possível entre todas as subtarefas da source. Importante Esta opção tem suporte apenas no 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 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 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 concluírem os próprios. Isso pode causar carga de trabalho desequilibrada 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 de shards e simplificar a alocação de tarefas.
  shardDiscoveryIntervalMs	O intervalo para descoberta dinâmica de shards.	Long	Não	60.000	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 é definido como true. Esta opção tem suporte apenas no VVR 8.0.9 e versões posteriores.
  startupMode	O modo de inicialização para a tabela 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, não há suporte para o valor consumer_group. 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 surtirá efeito.
  startTime	O horário de início para o consumo de log.	String	Não	Horário atual	O formato é yyyy-MM-dd hh:mm:ss. Isso só entra em vigor quando startupMode é 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	O horário de término para o consumo de log.	String	Não	Nenhum	O formato é yyyy-MM-dd hh:mm:ss. Nota Use esta opção apenas para consumir logs históricos e defina-a com um horário no passado. Se você a definir com um horário futuro, o consumo poderá parar prematuramente caso nenhum novo log seja gravado, resultando em interrupção no fluxo de dados sem mensagens de erro. Para que o job do Flink seja encerrado após o consumo de todos os logs, defina também exitAfterFinish como true.
  consumerGroup	O nome do grupo de consumidores.	String	Não	Nenhum	Use um grupo de consumidores para registrar o progresso do consumo. Especifique um nome personalizado sem formato fixo. Nota Jobs diferentes do Flink devem usar grupos de consumidores distintos. Se vários jobs do 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 o job deve consumir a partir de um checkpoint do grupo de consumidores.	String	Não	false	true: Especifique também um grupo de consumidores. O programa Flink começa a consumir logs a partir do checkpoint salvo 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 de startTime. false (valor padrão): Não inicia o consumo de logs a partir do checkpoint salvo para o grupo de consumidores especificado. Importante Não há mais suporte para este parâmetro no VVR 11.1 e versões posteriores. Para essas versões, 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	A configuração batchGetSize não pode exceder 1.000. Caso contrário, o sistema relatará um erro.
  exitAfterFinish	Especifica se o job do Flink é encerrado após o consumo de todos os dados.	String	Não	false	true: O programa Flink é encerrado após o consumo de todos os dados. 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 de o Flink consumi-los. Isso reduz custos e melhora a velocidade de processamento. Por exemplo, 'query' = '*| where request_method = ''GET''' indica que, antes de o Flink ler os dados do SLS, ele primeiro corresponde aos dados em que o valor do campo request_method é 'GET'. Nota Esta opção usa a linguagem SPL do Log Service (SLS). Para mais informações, consulte SPL syntax. Importante Esta opção tem suporte apenas no VVR 8.0.1 e versões posteriores. Este recurso incorre em taxas do Log Service (SLS). Para mais informações, consulte Billing of Log Service.
  processor	O nome do processador do SLS para pré-processamento de dados. Se você especificar tanto query quanto processor, query terá precedência e processor será ignorado.	String	Não	Nenhum	Esta opção filtra os dados do SLS antes de o Flink consumi-los, 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 do 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, consulte SPL syntax. Para obter informações sobre como criar ou atualize um processador do SLS, consulte Manage processors. Importante Esta opção tem suporte apenas no VVR 11.3 e versões posteriores. Este recurso incorre em taxas do Log Service (SLS). Para mais informações, consulte Billing of Log Service.
  preserveRawBytes	Especifica se o job deve buscar e preservar diretamente os bytes brutos transportados pelo SLS quando enableNewSource é definido 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 continuam a usar a lógica de conversão padrão. O comportamento é consistente com o da source antiga. Nota Esta opção tem suporte apenas no VVR 11.8 e versões posteriores. Esta opção só entra em vigor quando enableNewSource é definido como true.

• Específicas do sink

  Parâmetro	Descrição	Tipo	Obrigatório	Padrão	Observações
  topicField	selecione um campo para sobrescrever 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 para sobrescrever o atributo __timestamp__, que indica 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 você não especificar esta opção, o sistema usará o horário atual.
  sourceField	Especifica um campo para sobrescrever o atributo __source__, que indica a source 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. O 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 você não especificar esta opção, cada registro será gravado aleatoriamente em um shard disponível.
  buckets	Se você especificar partitionField, esta opção define 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	2.000	Unidade: milissegundos.
  writeNullProperties	Especifica se o sistema 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: O sistema não grava campos avaliados como nulos no log. Nota Esta opção tem suporte apenas no VVR 8.0.6 e versões posteriores.

Além dos campos de log e dos campos personalizados, o Realtime Compute for Apache Flink pode extrair os seguintes campos de atributo.

Para extrair campos de atributo, defina cabeçalhos na sua instrução SQL. Exemplo:

create table sls_stream(
  __timestamp__ bigint HEADER,
  __receive_time__ bigint HEADER
  b int,
  c varchar
) with (
  'connector' = 'sls',
  'endpoint' ='cn-hangzhou.log.aliyuncs.com',
  'accessId' = '${secret_values.ak_id}',
  'accessKey' = '${secret_values.ak_secret}',
  'starttime' = '2023-08-30 00:00:00',
  'project' ='sls-test',
  'logstore' ='sls-input'
);