Realtime Compute for Apache Flink:Log Service (SLS)

Le tableau suivant répertorie les fonctionnalités du connecteur SLS.

Le connecteur source SLS lit directement les champs d'attribut de message. Le tableau suivant répertorie les champs pris en charge.

Syntaxe

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

Avec options

• Général

  Paramètre	Description	Type	Obligatoire	Valeur par défaut	Remarques
  connector	Le connecteur à utiliser.	String	Oui	Aucun	Définissez ce paramètre sur sls.
  endPoint	L'endpoint de Log Service (SLS).	String	Oui	Aucun	Spécifiez l'adresse d'accès VPC de Log Service (SLS). Pour plus d'informations, consultez la rubrique Endpoints de service. Remarque Par défaut, Realtime Compute for Apache Flink ne peut pas accéder à Internet. Pour activer l'accès à Internet depuis votre Virtual Private Cloud (VPC), utilisez une passerelle NAT. Pour plus d'informations, consultez la rubrique Comment accéder à Internet ?. Nous vous déconseillons d'accéder à SLS via Internet. Si cela est indispensable, utilisez HTTPS et activez l'accélération du transfert. Pour plus d'informations, consultez la rubrique Gérer l'accélération du transfert.
  project	Le nom du projet SLS.	String	Oui	Aucun	S/O
  logStore	Le nom du Logstore ou du MetricStore.	String	Oui	Aucun	Les données d'un Logstore sont consommées de la même manière que celles d'un MetricStore.
  accessId	L'AccessKey ID de votre compte Alibaba Cloud.	String	Oui	Aucun	Pour plus d'informations, consultez la rubrique Obtenir une paire de clés AccessKey. Important Pour éviter l'exposition de votre paire de clés AccessKey, nous vous recommandons d'utiliser des variables pour spécifier l'AccessKey ID et l'AccessKey secret. Pour plus d'informations, consultez la rubrique Variables de projet.
  accessKey	L'AccessKey secret de votre compte Alibaba Cloud.	String	Oui	Aucun
  akless.assume-role.role-name	Le nom du rôle RAM utilisé pour accéder à SLS.	String	Non	Aucun	Si les paramètres accessId et accessKey ne sont pas spécifiés, vous devez définir ce paramètre pour accéder aux ressources SLS à l'aide d'un rôle RAM. Le rôle RAM doit être référencé au préalable sous Access Security/RAM Roles. Remarque Ce paramètre est pris en charge uniquement dans les versions VVR 11.9.0 et ultérieures.

• Spécifique à la source

  Paramètre	Description	Type	Obligatoire	Valeur par défaut	Remarques
  enableNewSource	Indique s'il faut utiliser la nouvelle source de données qui implémente l'interface FLIP-27.	Boolean	Non	false	La nouvelle source peut s'adapter automatiquement aux modifications des shards et répartir ces derniers de manière aussi équitable que possible entre toutes les sous-tâches de la source. Important Cette option est prise en charge uniquement dans les versions VVR 8.0.9 et ultérieures. La valeur par défaut est true pour les versions VVR 11.1 et ultérieures. Si vous modifiez la valeur de cette option, vous ne pouvez pas restaurer les tâches à partir d'un état enregistré. Pour reprendre la consommation à partir d'un offset historique, commencez par démarrer la tâche avec l'option consumerGroup afin d'enregistrer la progression de la consommation dans le groupe de consommateurs SLS. Ensuite, définissez l'option consumeFromCheckpoint sur true et redémarrez la tâche sans état. Si un Logstore contient des shards en lecture seule, certaines sous-tâches peuvent continuer à demander des données provenant d'autres shards après avoir terminé leur propre traitement. Cela peut entraîner un déséquilibre de la charge de travail, affectant ainsi les performances. Pour atténuer ce problème, vous pouvez ajuster le parallélisme, optimiser la stratégie de planification ou fusionner les petits shards afin de réduire leur nombre et de simplifier l'allocation des tâches.
  shardDiscoveryIntervalMs	L'intervalle pour la découverte dynamique des shards.	Long	Non	60000	Définissez cette option sur une valeur négative pour désactiver la découverte dynamique. Unité : millisecondes. Remarque La valeur doit être supérieure ou égale à 60 000 millisecondes (1 minute). Cette option prend effet uniquement lorsque enableNewSource est défini sur true. Cette option est prise en charge uniquement dans les versions VVR 8.0.9 et ultérieures.
  startupMode	Le mode de démarrage pour la table source.	String	Non	timestamp	timestamp (par défaut) : Consomme les journaux à partir de l'heure de début spécifiée. latest : Commence à consommer les journaux à partir du dernier offset. earliest : Commence à consommer les journaux à partir du premier offset. consumer_group : Commence la consommation des journaux à partir de l'offset enregistré par le groupe de consommateurs. Si le groupe de consommateurs n'a enregistré aucun offset de consommation pour un shard, la consommation commence à partir du premier offset. Important Pour les versions VVR antérieures à 11,1, la valeur consumer_group n'est pas prise en charge. Vous devez définir consumeFromCheckpoint sur true. Dans ce cas, la consommation des journaux commence à partir de l'offset enregistré par le groupe de consommateurs spécifié, et le paramètre de mode de démarrage ne prend pas effet.
  startTime	L'heure de début de la consommation des journaux.	String	Non	Heure actuelle	Le format est yyyy-MM-dd hh:mm:ss. Cela prend effet uniquement lorsque startupMode est défini sur timestamp. Remarque Les options startTime et stopTime sont basées sur l'attribut __receive_time__ dans SLS, plutôt que sur l'attribut __timestamp__.
  stopTime	L'heure de fin de la consommation des journaux.	String	Non	Aucun	Le format est yyyy-MM-dd hh:mm:ss. Remarque Cette option est utilisée uniquement pour la consommation de journaux historiques et doit être définie sur une heure passée. Si vous la définissez sur une heure future, la consommation risque de s'arrêter prématurément si aucun nouveau journal n'est écrit, ce qui entraîne une interruption du flux de données sans message d'erreur. Si vous souhaitez que la tâche Flink se termine après la consommation de tous les journaux, vous devez également définir exitAfterFinish sur true.
  consumerGroup	Le nom du groupe de consommateurs.	String	Non	Aucun	Un groupe de consommateurs sert à enregistrer la progression de la consommation. Vous pouvez spécifier un nom personnalisé sans format fixe. Remarque Différentes tâches Flink doivent utiliser des groupes de consommateurs distincts. Si plusieurs tâches Flink utilisent le même groupe de consommateurs, elles ne se coordonnent pas et chaque tâche consomme toutes les données. En effet, Flink n'utilise pas le groupe de consommateurs SLS pour l'attribution des partitions lors de la consommation des données depuis SLS. Par conséquent, chaque consommateur traite les messages indépendamment, même s'ils partagent le même groupe de consommateurs.
  consumeFromCheckpoint	Indique s'il faut consommer à partir d'un point de contrôle du groupe de consommateurs.	String	Non	false	true : Vous devez également spécifier un groupe de consommateurs. Le programme Flink commence à consommer les journaux à partir du point de contrôle enregistré dans le groupe de consommateurs. Si le groupe de consommateurs ne dispose pas de point de contrôle correspondant, la consommation commence à partir de la valeur configurée pour startTime. false (valeur par défaut) : Ne commence pas à consommer les journaux à partir du point de contrôle enregistré pour le groupe de consommateurs spécifié. Important Ce paramètre n'est plus pris en charge dans les versions VVR 11.1 et ultérieures. Pour ces versions, vous devez définir l'option startupMode sur consumer_group.
  maxRetries	Le nombre de tentatives après un échec de lecture depuis SLS.	String	Non	3	S/O
  batchGetSize	Le nombre de groupes de journaux à lire en une seule requête.	String	Non	100	Le paramètre batchGetSize ne peut pas dépasser 1 000, sinon une erreur est signalée.
  exitAfterFinish	Indique si la tâche Flink se termine après la consommation de toutes les données.	String	Non	false	true : Le programme Flink se termine après la consommation de toutes les données. false (par défaut) : Le programme Flink ne se termine pas après la consommation des données.
  query Important Cette option a été dépréciée dans la version VVR 11.3, mais les versions ultérieures restent compatibles.	L'instruction de requête pour le prétraitement des données avant la consommation.	String	Non	Aucun	Utilisez cette option pour filtrer les données SLS avant leur consommation par Flink. Cela permet de réduire les coûts et d'améliorer la vitesse de traitement. Par exemple, 'query' = '*| where request_method = ''GET''' indique qu'avant que Flink ne lise les données depuis SLS, il filtre d'abord les données dont la valeur du champ request_method est « GET ». Remarque Cette option utilise le langage SPL de Log Service (SLS). Pour plus d'informations, consultez la rubrique Syntaxe SPL. Important Cette option est prise en charge uniquement dans les versions VVR 8.0.1 et ultérieures. Cette fonctionnalité engendre des frais Log Service (SLS). Pour plus d'informations, consultez la rubrique Facturation de Log Service.
  processor	Le nom du processeur SLS pour le prétraitement des données. Si les paramètres query et processor sont tous deux spécifiés, query est prioritaire et processor est ignoré.	String	Non	Aucun	Cette option filtre les données SLS avant leur consommation par Flink, ce qui réduit les coûts et améliore la vitesse de traitement. Nous vous recommandons d'utiliser processor plutôt que query. Par exemple, 'processor' = 'test-filter-processor' indique que le processeur SLS filtre les données avant que Flink ne les lise depuis SLS. Remarque Cette option utilise le langage SPL de Log Service (SLS). Pour plus d'informations, consultez la rubrique Syntaxe SPL. Pour savoir comment créer ou mettre à jour un processeur SLS, consultez la rubrique Gérer les processeurs. Important Cette option est prise en charge uniquement dans les versions VVR 11.3 et ultérieures. Cette fonctionnalité engendre des frais Log Service (SLS). Pour plus d'informations, consultez la rubrique Facturation de Log Service.
  preserveRawBytes	Indique s'il faut récupérer et conserver directement les octets bruts transportés par SLS lorsque enableNewSource est défini sur true.	Boolean	Non	false	Lorsque cette option est activée, les champs BINARY/VARBINARY lisent directement le tableau d'octets brut (byte[]), tandis que les champs CHAR/VARCHAR construisent les données de chaîne à partir des octets bruts. Les autres champs utilisent toujours la logique de conversion par défaut. Le comportement est cohérent avec celui de l'ancienne source. Remarque Cette option est prise en charge uniquement dans les versions VVR 11.8 et ultérieures. Cette option prend effet uniquement lorsque enableNewSource est défini sur true.

• Spécifique au sink

  Paramètre	Description	Type	Obligatoire	Valeur par défaut	Remarques
  topicField	Spécifie un champ dont la valeur remplace l'attribut __topic__, qui indique le sujet du journal.	String	Non	Aucun	La valeur de cette option doit correspondre à un champ existant dans la table.
  timeField	Spécifie un champ dont la valeur remplace l'attribut __timestamp__, qui indique l'heure d'écriture du journal.	String	Non	Heure actuelle	La valeur de cette option doit correspondre à un champ INT existant dans la table. Si cette option n'est pas spécifiée, l'heure actuelle est utilisée.
  sourceField	Spécifie un champ dont la valeur remplace l'attribut __source__, qui indique la source du journal, telle que l'adresse IP de la machine ayant généré le journal.	String	Non	Aucun	La valeur de cette option doit correspondre à un champ existant dans la table.
  partitionField	Spécifie un champ pour le partitionnement. Un hachage de la valeur de ce champ détermine le shard qui reçoit les données, garantissant que les enregistrements ayant le même hachage sont dirigés vers le même shard.	String	Non	Aucun	Si cette option n'est pas spécifiée, chaque enregistrement est écrit aléatoirement dans un shard disponible.
  buckets	Si partitionField est spécifié, cette option définit le nombre de buckets pour le mappage des valeurs de hachage.	String	Non	64	La valeur doit être une puissance de 2 comprise dans l'intervalle [1, 256]. Le nombre de buckets doit être supérieur ou égal au nombre de shards. Sinon, certains shards risquent de ne recevoir aucune donnée.
  flushIntervalMs	L'intervalle d'écriture des données.	String	Non	2000	Unité : millisecondes.
  writeNullProperties	Indique s'il faut écrire les valeurs nulles sous forme de chaînes vides dans SLS.	Boolean	Non	true	true (valeur par défaut) : Écrit les valeurs nulles dans le journal sous forme de chaînes vides. false : Les champs évalués comme nuls ne sont pas écrits dans le journal. Remarque Cette option est prise en charge uniquement dans les versions VVR 8.0.6 et ultérieures.

Correspondances de types

Limites

Cette fonctionnalité est prise en charge uniquement par les versions 11.1 et ultérieures de Realtime Compute for Apache Flink.

Syntaxe

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

Paramètres

Réutilisation d'un catalogue existant

À partir de la version 11.5 de Realtime Compute for Apache Flink, vous pouvez référencer un catalogue SLS intégré créé sur la page Data Management dans les tâches d'ingestion de données Flink CDC. Cette approche évite de devoir spécifier manuellement les propriétés de connexion.

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

Actuellement, les tâches d'ingestion de données peuvent réutiliser automatiquement les paramètres suivants issus du catalogue SLS intégré :

• endpoint

• project

• accessId

• accessKey

Pour remplacer ces paramètres réutilisés automatiquement, définissez-les explicitement dans la configuration YAML. Les paramètres définis dans le fichier YAML sont prioritaires.

Correspondance des types de données

Si fixed-types n'est pas configuré, la correspondance des types de données suivante s'applique :

Si fixed-types est configuré, le système analyse les données en utilisant les types spécifiés.

Inférence et évolution du schéma

• Pré-consommation des données des shards et initialisation du schéma

  Le connecteur SLS conserve le schéma du Logstore à partir duquel il lit les données. Avant de lire les données du Logstore, le connecteur pré-consomme jusqu'à maxPreFetchLogGroups groupes de journaux depuis chaque shard. Il analyse le schéma de chaque entrée de journal et les fusionne pour initialiser le schéma de la table. Un événement de création de table est ensuite généré sur la base de ce schéma initial avant le début de la consommation des données.

  Remarque

  Pour chaque shard, le connecteur tente de consommer les données à partir d'une heure avant l'heure actuelle afin d'analyser le schéma des journaux.

• Informations sur la clé primaire

  Les journaux Log Service (SLS) ne contiennent pas d'informations sur la clé primaire. Vous pouvez ajouter manuellement une clé primaire à la table à l'aide de règles de transformation :

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

• Inférence du schéma et modifications du schéma

  Une fois le schéma initialisé, si schema.inference.strategy est défini sur static, le connecteur SLS analyse chaque entrée de journal selon le schéma initial et ne génère pas d'événements de modification du schéma. Si schema.inference.strategy est défini sur continuous, le connecteur analyse chaque entrée de journal, déduit les colonnes physiques et les compare au schéma actuel. Si le schéma déduit est incohérent avec le schéma actuel, les schémas sont fusionnés selon les règles suivantes :

  • Si les colonnes physiques déduites contiennent des champs absents du schéma actuel, le connecteur ajoute ces champs au schéma et génère un événement pour ajouter des colonnes acceptant les valeurs NULL.

  • Si les colonnes physiques déduites omettent des champs présents dans le schéma actuel, le connecteur conserve ces champs, remplit leurs données par NULL et ne génère pas d'événement de suppression de colonne.

  Le connecteur SLS déduit le type de données de tous les champs de chaque entrée de journal comme étant String. Actuellement, seule l'ajout de nouvelles colonnes est pris en charge. Le connecteur ajoute les nouvelles colonnes à la fin du schéma et les définit comme acceptant les valeurs NULL.

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