Realtime Compute for Apache Flink:Log Service (SLS)

A continuación se muestran las capacidades del conector de SLS.

El conector de origen de SLS lee directamente los campos de atributos de los mensajes. La siguiente tabla muestra los campos compatibles.

Sintaxis

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

Opciones WITH

• Generales

  Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
  connector	El conector que se va a utilizar.	String	Sí	Ninguna	Establezca este valor en sls.
  endPoint	El endpoint de Log Service (SLS).	String	Sí	Ninguna	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 tiene acceso 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 es indispensable 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í	Ninguna	N/D
  logStore	El nombre del logstore o MetricStore.	String	Sí	Ninguna	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í	Ninguna	Para obtener más información, consulte Obtain an AccessKey pair. Importante Para evitar exponer su par de AccessKey, se recomienda usar 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í	Ninguna
  akless.assume-role.role-name	El nombre del rol de RAM que se utiliza para acceder a SLS.	String	No	Ninguna	Si no se especifican accessId y accessKey, debe configurar 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 forma más equitativa 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, primero inicie 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 solo surte efecto 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 registros a partir de la hora de inicio especificada. latest: inicia el consumo de registros desde el desplazamiento más reciente. earliest: inicia el consumo de 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. Solo surte efecto 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	Ninguna	El formato es yyyy-MM-dd hh:mm:ss. Nota Esta opción solo se utiliza para consumir registros históricos y debe configurarse con una hora pasada. Si la establece en una hora futura, el consumo podría detenerse de forma prematura cuando no se escriban nuevos registros. 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	Ninguna	Un grupo de consumidores sirve para registrar el progreso de consumo. Puede especificar un nombre personalizado sin un formato fijo. Nota Cada trabajo de Flink debe usar un grupo de consumidores distinto. 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 SLS. 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 registros desde el punto de control guardado en dicho grupo. Si el grupo de consumidores no tiene un punto de control correspondiente, el consumo 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 estas versiones, debe establecer la opción startupMode en consumer_group.
  maxRetries	El número de reintentos tras un intento fallido de lectura desde 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	Ninguna	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 leer datos de SLS, Flink selecciona aquellos donde el valor del campo request_method es '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	Ninguna	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 usar 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 desde SLS. 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 esos 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 solo surte efecto 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	Ninguna	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	Ninguna	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 envíen al mismo shard.	String	No	Ninguna	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 cantidad de shards. De lo contrario, algunos shards podrían no recibir 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.

Asignación de tipos

Limitaciones

Esta función solo es compatible con Realtime Compute for Apache Flink versión 11.1 y posteriores.

Sintaxis

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

Parámetros

Reutilizar un catálogo existente

A partir de la versión 11.5 de Realtime Compute for Apache Flink, puede hacer referencia a un SLS Catalog integrado creado en la página Data Management dentro de los trabajos de ingesta de datos de Flink CDC. Esto evita tener que especificar manualmente las propiedades de conexión.

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

Actualmente, los trabajos de ingesta de datos pueden reutilizar automáticamente los siguientes parámetros del SLS Catalog integrado:

• endpoint

• project

• accessId

• accessKey

Para anular estos parámetros reutilizados automáticamente, defínalos explícitamente en la configuración YAML. Los parámetros definidos en el archivo YAML tienen prioridad.

Mapeo de tipos de datos

Si no se configura fixed-types, se aplica el siguiente mapeo de tipos de datos:

Si se configura fixed-types, el sistema analiza los datos con los tipos especificados.

Inference y evolución del esquema

• Preconsumo de datos de fragmentos e inicialización del esquema

  El conector de SLS mantiene el esquema del logstore del que lee. Antes de leer los datos del logstore, el conector preconsume hasta maxPreFetchLogGroups grupos de registros de cada fragmento. Analiza el esquema de cada entrada de registro y los combina para inicializar el esquema de la tabla. Luego, genera un evento de creación de tabla con base en este esquema inicial antes de comenzar el consumo de datos.

  Nota

  Para cada fragmento, el conector intenta consumir datos comenzando una hora antes de la hora actual para analizar el esquema del registro.

• Información de clave primaria

  Los registros de Log Service (SLS) no contienen información de clave primaria. Agregue manualmente una clave primaria a la tabla mediante reglas de transformación:

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

• Inference y cambios de esquema

  Una vez inicializado el esquema, si schema.inference.strategy se establece en static, el conector de SLS analiza cada entrada de registro con base en el esquema inicial y no genera eventos de cambio de esquema. Si schema.inference.strategy se establece en continuous, el conector analiza cada entrada de registro, deduce las columnas físicas y las compara con el esquema actual. Si el esquema inferido no coincide con el actual, los esquemas se combinan según las siguientes reglas:

  • Si las columnas físicas inferidas contienen campos que no están en el esquema actual, el conector agrega estos campos al esquema y genera un evento para añadir columnas anulables.

  • Si a las columnas físicas inferidas les faltan campos que existen en el esquema actual, el conector conserva dichos campos, llena sus datos con NULL y no genera un evento de eliminación de columnas.

  El conector de SLS deduce el tipo de datos de todos los campos en cada entrada de registro como String. Actualmente, solo se admite agregar nuevas columnas. El conector añade las columnas nuevas al final del esquema y las configura como anulables.

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