Simple Log Service:Consumo de datos de registro con Realtime Compute for Apache Flink

Simple Log Service almacena los datos de registro en tiempo real y Realtime Compute for Apache Flink los lee en modo de streaming. A continuación se muestra un ejemplo de entrada de registro:

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

Código de ejemplo

En el siguiente ejemplo de borrador SQL se consumen datos de Simple Log Service en 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; 

Opciones WITH

• Generales

  Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
  connector	El conector que se va a utilizar.	String	Sí	Ninguno	Establezca este valor en sls.
  endPoint	El endpoint de Log Service (SLS).	String	Sí	Ninguno	Especifique la dirección de acceso de VPC de Log Service (SLS). Para obtener más información, consulte Service endpoints. Nota De forma predeterminada, Realtime Compute for Apache Flink no puede acceder a Internet. Para habilitar el acceso a Internet desde su Virtual Private Cloud (VPC), utilice una NAT Gateway. Para obtener más información, consulte How do I access the Internet?. Se recomienda no acceder a SLS a través de Internet. Si debe hacerlo, utilice HTTPS y habilite la aceleración de transferencia. Para obtener más información, consulte Manage transfer acceleration.
  project	El nombre del proyecto de SLS.	String	Sí	Ninguno	N/D
  logStore	El nombre del logstore o MetricStore.	String	Sí	Ninguno	Los datos de un logstore se consumen de la misma manera que los de un MetricStore.
  accessId	El AccessKey ID de su cuenta de Alibaba Cloud.	String	Sí	Ninguno	Para obtener más información, consulte Obtain an AccessKey pair. Importante Para evitar que su par de AccessKey quede expuesto, se recomienda utilizar variables para especificar el AccessKey ID y el AccessKey secret. Para obtener más información, consulte Project variables.
  accessKey	El AccessKey secret de su cuenta de Alibaba Cloud.	String	Sí	Ninguno
  akless.assume-role.role-name	El nombre del rol de RAM que se utiliza para acceder a SLS.	String	No	Ninguno	Si no se especifican accessId y accessKey, debe indicar este parámetro para acceder a los recursos de SLS mediante un rol de RAM. El rol de RAM debe referenciarse previamente en Access Security/RAM Roles. Nota Este parámetro solo es compatible con VVR 11.9.0 y versiones posteriores.

• Específicas del origen

  Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
  enableNewSource	Indica si se utiliza el nuevo origen de datos que implementa la interfaz FLIP-27.	Boolean	No	false	El nuevo origen se adapta automáticamente a los cambios de shards y los distribuye de la manera más uniforme posible entre todas las subtareas de origen. Importante Esta opción solo es compatible con VVR 8.0.9 y versiones posteriores. El valor predeterminado es true en VVR 11.1 y versiones posteriores. Si modifica el valor de esta opción, no podrá restaurar trabajos desde un estado guardado. Para reanudar el consumo desde un desplazamiento histórico, inicie primero el trabajo con la opción consumerGroup para registrar el progreso de consumo en el grupo de consumidores de SLS. Luego, establezca la opción consumeFromCheckpoint en true y reinicie el trabajo sin estado. Si un logstore tiene shards de solo lectura, algunas subtareas podrían seguir solicitando datos de otros shards después de terminar los suyos. Esto genera una carga de trabajo desequilibrada que afecta el rendimiento. Para mitigar este problema, ajuste el paralelismo, optimice la estrategia de programación o combine shards pequeños para reducir su cantidad y simplificar la asignación de tareas.
  shardDiscoveryIntervalMs	El intervalo para la detección dinámica de shards.	Long	No	60000	Establezca esta opción en un valor negativo para deshabilitar la detección dinámica. Unidad: milisegundos. Nota El valor debe ser mayor o igual a 60 000 milisegundos (1 minuto). Esta opción surte efecto solo cuando enableNewSource se establece en true. Esta opción solo es compatible con VVR 8.0.9 y versiones posteriores.
  startupMode	El modo de inicio de la tabla de origen.	String	No	timestamp	timestamp (predeterminado): consume los registros a partir de la hora de inicio especificada. latest: comienza a consumir los registros desde el desplazamiento más reciente. earliest: comienza a consumir los registros desde el desplazamiento más antiguo. consumer_group: inicia el consumo de registros desde el desplazamiento registrado por el grupo de consumidores. Si el grupo de consumidores no ha registrado un desplazamiento de consumo para un shard, el consumo comienza desde el desplazamiento más antiguo. Importante En versiones de VVR anteriores a la 11.1, no se admite el valor consumer_group. Debe establecer consumeFromCheckpoint en true. En este caso, el consumo de registros comienza desde el desplazamiento registrado por el grupo de consumidores especificado y la configuración del modo de inicio no surte efecto.
  startTime	La hora de inicio para el consumo de registros.	String	No	Hora actual	El formato es yyyy-MM-dd hh:mm:ss. Surte efecto solo cuando startupMode se establece en timestamp. Nota Las opciones startTime y stopTime se basan en el atributo __receive_time__ de SLS, no en el atributo __timestamp__.
  stopTime	La hora de finalización para el consumo de registros.	String	No	Ninguno	El formato es yyyy-MM-dd hh:mm:ss. Nota Esta opción se usa únicamente para consumir registros históricos y debe configurarse con una hora pasada. Si la establece en una hora futura, el consumo podría detenerse antes de tiempo cuando no se escriban registros nuevos. Esto provoca una interrupción del flujo de datos sin mensajes de error. Si desea que el trabajo de Flink termine después de consumir todos los registros, también debe establecer exitAfterFinish en true.
  consumerGroup	El nombre del grupo de consumidores.	String	No	Ninguno	Un grupo de consumidores sirve para registrar el progreso de consumo. Especifique un nombre personalizado sin un formato fijo. Nota Los distintos trabajos de Flink deben usar grupos de consumidores diferentes. Si varios trabajos de Flink utilizan el mismo grupo de consumidores, no se coordinan entre sí y cada uno consume todos los datos. Esto ocurre porque Flink no emplea el grupo de consumidores de SLS para asignar particiones cuando consume datos de este servicio. Como resultado, cada consumidor procesa los mensajes de forma independiente, incluso si comparten el mismo grupo de consumidores.
  consumeFromCheckpoint	Indica si se consume desde un punto de control del grupo de consumidores.	String	No	false	true: también debe especificar un grupo de consumidores. El programa de Flink comienza a consumir los registros desde el punto de control guardado en dicho grupo. Si el grupo de consumidores no tiene un punto de control correspondiente, el consumo se inicia desde el valor de configuración de startTime. false (valor predeterminado): no inicia el consumo de registros desde el punto de control guardado para el grupo de consumidores especificado. Importante Este parámetro ya no es compatible con VVR 11.1 y versiones posteriores. En dichas versiones, debe establecer la opción startupMode en consumer_group.
  maxRetries	El número de reintentos tras un intento fallido de lectura de SLS.	String	No	3	N/D
  batchGetSize	El número de grupos de registros que se leen en una sola solicitud.	String	No	100	La configuración de batchGetSize no puede superar 1000; de lo contrario, se reporta un error.
  exitAfterFinish	Indica si el trabajo de Flink termina después de consumir todos los datos.	String	No	false	true: el programa de Flink termina después de consumir todos los datos. false (predeterminado): el programa de Flink no termina al finalizar el consumo de datos.
  query Importante Esta opción quedó obsoleta en VVR 11.3, pero las versiones posteriores siguen siendo compatibles.	La instrucción de consulta para preprocesar los datos antes del consumo.	String	No	Ninguno	Utilice esta opción para filtrar los datos de SLS antes de que Flink los consuma. Esto reduce costos y mejora la velocidad de procesamiento. Por ejemplo, 'query' = '*| where request_method = ''GET''' indica que, antes de que Flink lea los datos de SLS, primero busca aquellos donde el valor del campo request_method sea 'GET'. Nota Esta opción utiliza el lenguaje SPL de Log Service (SLS). Para obtener más información, consulte SPL syntax. Importante Esta opción solo es compatible con VVR 8.0.1 y versiones posteriores. Esta función genera cargos de Log Service (SLS). Para obtener más información, consulte Billing of Log Service.
  processor	El nombre del procesador de SLS para el preprocesamiento de datos. Si se especifican tanto query como processor, query tiene prioridad y processor se ignora.	String	No	Ninguno	Esta opción filtra los datos de SLS antes de que Flink los consuma, lo cual reduce costos y mejora la velocidad de procesamiento. Se recomienda utilizar processor en lugar de query. Por ejemplo, 'processor' = 'test-filter-processor' indica que el procesador de SLS filtra los datos antes de que Flink los lea. Nota Esta opción utiliza el lenguaje SPL de Log Service (SLS). Para obtener más información, consulte SPL syntax. Para conocer cómo crear o actualizar un procesador de SLS, consulte Manage processors. Importante Esta opción solo es compatible con VVR 11.3 y versiones posteriores. Esta función genera cargos de Log Service (SLS). Para obtener más información, consulte Billing of Log Service.
  preserveRawBytes	Indica si se obtienen y conservan directamente los bytes sin procesar transportados por SLS cuando enableNewSource se establece en true.	Boolean	No	false	Cuando esta opción está habilitada, los campos BINARY/VARBINARY leen directamente el byte[] sin procesar y los campos CHAR/VARCHAR construyen los datos de cadena a partir de dichos bytes. Los demás campos siguen utilizando la lógica de conversión predeterminada. Este comportamiento coincide con el del origen anterior. Nota Esta opción solo es compatible con VVR 11.8 y versiones posteriores. Esta opción surte efecto solo cuando enableNewSource se establece en true.

• Específicas del destino

  Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
  topicField	Especifica un campo cuyo valor sobrescribe el atributo __topic__, que indica el tema del registro.	String	No	Ninguno	El valor de esta opción debe ser un campo existente en la tabla.
  timeField	Especifica un campo cuyo valor sobrescribe el atributo __timestamp__, que indica la hora de escritura del registro.	String	No	Hora actual	El valor de esta opción debe ser un campo INT existente en la tabla. Si no se especifica, se utiliza la hora actual.
  sourceField	Especifica un campo cuyo valor sobrescribe el atributo __source__, que indica el origen del registro, como la dirección IP de la máquina que lo generó.	String	No	Ninguno	El valor de esta opción debe ser un campo existente en la tabla.
  partitionField	Especifica un campo para la creación de particiones. Un hash del valor de este campo determina qué shard recibe los datos, lo que garantiza que los registros con el mismo hash se dirijan al mismo shard.	String	No	Ninguno	Si no se especifica esta opción, cada registro se escribe de forma aleatoria en un shard disponible.
  buckets	Si se especifica partitionField, esta opción define el número de buckets para asignar los valores de hash.	String	No	64	El valor debe ser una potencia de 2 dentro del rango [1, 256]. La cantidad de buckets debe ser mayor o igual que la de shards. De lo contrario, es posible que algunos shards no reciban datos.
  flushIntervalMs	El intervalo de escritura de datos.	String	No	2000	Unidad: milisegundos.
  writeNullProperties	Indica si los valores nulos se escriben en SLS como cadenas vacías.	Boolean	No	true	true (valor predeterminado): escribe los valores nulos en el registro como cadenas vacías. false: los campos que se evalúan como nulos no se escriben en el registro. Nota Esta opción solo es compatible con VVR 8.0.6 y versiones posteriores.

Además de los campos de registro y los campos personalizados, Realtime Compute for Apache Flink puede extraer los siguientes campos de atributos.

Para extraer campos de atributos, defina encabezados en su instrucción SQL. Ejemplo:

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'
);