Realtime Compute for Apache Flink:Connecteurs

• Kafka

  • Comment récupérer des données JSON depuis Kafka

  • Pourquoi Flink ne peut-il pas consommer ou écrire dans Kafka malgré une connexion réseau fonctionnelle ?

  • Absence de sortie de données d'une source Kafka avec fenêtre temporelle événementielle

  • Objectif de la validation des offsets dans Kafka

  • Analyse de données JSON imbriquées avec le connecteur Kafka

  • Connexion à un cluster Kafka sécurisé

  • Résolution des conflits de noms de champs

  • Gestion d'une latence inattendue d'une source Kafka

  • Erreur : 'upsert-kafka' table require PRIMARY KEY

• DataHub

  • Reprise du job après la mise à l'échelle d'un topic DataHub

  • Puis-je supprimer un topic DataHub déjà consommé ?

  • Échec de création de table DataHub avec NoPermissionException: dhs:ListShard

• MaxCompute

  • Comprendre endPoint et tunnelEndpoint

  • Lecture des données pour les sources MaxCompute complètes et incrémentielles

  • Les sources MaxCompute peuvent-elles lire les données ajoutées après le démarrage du job ?

  • Modification de la concurrence pour les jobs source MaxCompute repris

  • Pourquoi une source MaxCompute lit-elle des partitions antérieures à la position de départ ?

  • Gestion des nouvelles partitions incomplètes dans une source MaxCompute incrémentielle

  • [Erreur du connecteur MaxCompute : ErrorMessage=Authorization Failed [4019], You have NO privilege](#section_947_odv_kbe)

  • Configuration du paramètre startPartition pour une source MaxCompute incrémentielle

  • Long délai avant qu'une source MaxCompute incrémentielle ne commence à lire

  • Configuration du paramètre Partition

  • Job avec source MaxCompute bloqué au démarrage ou retardé

  • Sélection d'un Tunnel

  • Gestion des données en double dans les tables de résultats MaxCompute

  • Erreur d'exécution dans la table de résultats MaxCompute : 'Invalid partition spec'

  • Erreur d'exécution dans la table de résultats MaxCompute : 'No more available blockId'

  • Différence entre max_pt() et max_pt_with_done()

  • Utilisation de l'indice SHUFFLE_HASH

  • Configuration du paramètre CacheReloadTimeBlackList

• MySQL

  • Erreur : java.io.EOFException: SSL peer shut down incorrectly

• ApsaraDB RDS for MySQL

  • Changements de type pour la clé primaire bigint unsigned lors de la synchronisation des données

  • Flink met-il à jour ou insère-t-il des enregistrements lors de l'écriture dans RDS ?

  • Utilisation de GROUP BY sur un index unique dans une table RDS

  • Pourquoi INT UNSIGNED dans MySQL nécessite un type différent dans Flink SQL

  • Erreur : Incorrect string value: '\xF0\x9F\x98\x80\xF0\x9F...' for column 'test' at row 1

  • Interblocage lors de l'écriture dans MySQL

  • Le schéma en aval ne change pas après la mise à jour du schéma MySQL

  • Causes de l'exception 'finish split response timeout'

  • Effets des changements de schéma lors de la synchronisation complète CDC MySQL

  • Résolution des échecs de job dus à des changements de schéma non pris en charge lors de la synchronisation CTAS/CDAS

• ClickHouse

  • Rétractation et mise à jour des données dans les tables de résultats ClickHouse

  • Visibilité des données dans les tables de résultats ClickHouse

• Print

  Consultation des résultats print dans la Console

• Tablestore

  La jointure JOIN sur une table de dimension ne renvoie aucune donnée

• ApsaraMQ for RocketMQ

  • Erreur : IllegalArgumentException: timeout value is negative

  • Comment RocketMQ détecte les changements de partition lors de la montée en charge du topic

• Hologres

  • Erreur : BackPressure Exceed reject Limit

  • Erreur : remaining connection slots are reserved for non-replication superuser connections

  • Erreur : no table is defined in publication

  • Relation entre l'intervalle de checkpoint et la visibilité des données pour le sink Hologres

  • Exception 'permission denied for database' lors du déploiement

  • Exception 'table id ...' lors de la restauration depuis un checkpoint

  • Problèmes de précision des données lors de la consommation de données binlog en mode JDBC

  • Erreurs après la suppression et la recréation d'une table portant le même nom

  • ClassCastException lors de la lecture depuis Hologres

• Log Service

  • Erreur : Caused by: com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException: the logs is 8785684 bytes which is larger than MAX_BATCH_SIZE_IN_BYTES 8388608

  • Erreur OOM sur le TaskManager lors de la restauration d'un job ayant échoué

• Paimon

  • Définition de la position de démarrage pour une source Paimon

  • Configuration de l'expiration automatique des partitions

  • Échec du job d'écriture Paimon avec 'Heartbeat of TaskManager timed out'

  • Échec du job d'écriture Paimon avec l'erreur 'Sink materializer must not be used with Paimon sink'

  • Erreurs de conflit lors de l'écriture dans Paimon

  • Échec du job de lecture Paimon avec l'erreur 'File xxx not found'

  • Échec du job Paimon avec 'No space left on device'

  • Gestion d'un grand nombre de fichiers Paimon sur OSS

  • Visibilité des données Paimon et intervalle de checkpoint

  • Augmentation lente de la mémoire dans les jobs écrivant vers Paimon

  • Décalage de fuseau horaire de 8 heures lorsqu'un job Flink SQL consomme des données Paimon et écrit dans Kafka

• Hudi

  • Aucune donnée trouvée dans le stockage

  • Gestion des données en double dans Hudi

  • Pourquoi une table Merge On Read ne contient que des fichiers journaux

• AnalyticDB for MySQL (ADB)

  Erreur : multi-statement be found

• Connecteur personnalisé

  • Erreur : No suitable driver found for

• Elasticsearch

  • Perte ou écrasement de données lorsque Flink écrit dans Elasticsearch

Récupérer des données JSON depuis Kafka avec Flink

• Pour récupérer des données JSON standard, consultez JSON Format.

• Pour récupérer des données JSON imbriquées, définissez l'objet JSON comme un type ROW dans la DDL de la table source. Dans la DDL de la table sink, déclarez les clés à récupérer. Utilisez ensuite une instruction DML pour accéder aux clés et extraire leurs valeurs. L'exemple de code suivant illustre cette procédure :

  • Exemple de données

    {
        "a":"abc",
        "b":1,
        "c":{
            "e":["1","2","3","4"],
            "f":{"m":"567"}
        }
    }

  • DDL de la table source

    CREATE TEMPORARY TABLE `kafka_table` (
      `a` VARCHAR,
       b int,
      `c` ROW<e ARRAY<VARCHAR>,f ROW<m VARCHAR>>  -- 'c' is a JSON object that maps to the ROW type in Flink. 'e' is a JSON array that maps to the ARRAY type.
    ) WITH (
      'connector' = 'kafka',
      'topic' = 'xxx',
      'properties.bootstrap.servers' = 'xxx',
      'properties.group.id' = 'xxx',
      'format' = 'json',
      'scan.startup.mode' = 'xxx'
    );

  • DDL de la table sink

    CREATE TEMPORARY TABLE `sink` (
     `a` VARCHAR,
      b INT,
      e VARCHAR,
      `m` varchar
    ) WITH (
      'connector' = 'print',
      'logger' = 'true'
    );

  • Instruction DML

    INSERT INTO `sink`
      SELECT 
      `a`,
      b,
      c.e[1], -- Flink uses 1-based indexing for arrays. This example uses index 1 to retrieve the first element. To retrieve the entire array, omit [1].
      c.f.m
    FROM `kafka_table`;

  • Résultats

    409  2021-04-08 10:13:11,214 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] -
    410  2021-04-08 10:13:11,215 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] - Kafka commitId: cb8625948210849f
    411  2021-04-08 10:13:11,215 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.common.utils.AppInfoParser [] - Kafka startTimeMs: 1617847991214
    412  2021-04-08 10:13:11,270 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Subscribed to partition(s): lb_test-0, lb_test-1, lb_test-2, lb_test-3, lb_test-4, lb_test-5
    413  2021-04-08 10:13:11,280 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 1 for partition lb_test-0
    414  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-1
    415  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-2
    416  2021-04-08 10:13:11,285 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-3
    417  2021-04-08 10:13:11,290 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-4
    418  2021-04-08 10:13:11,291 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.consumer.KafkaConsumer [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Seeking to offset 0 for partition lb_test-5
    419  2021-04-08 10:13:11,302 INFO  org.apache.flink.kafka.shaded.org.apache.kafka.clients.Metadata [] - [Consumer clientId=consumer-lb_test-2, groupId=lb_test] Cluster ID: -1flJPwnTvuGFSuyCtU1hw
    420  2021-04-08 10:15:31,597 INFO  org.apache.flink.api.common.functions.util.PrintSinkOutputWriter [] - +I(abc,1,1,567)

Flink ne peut pas consommer ou écrire dans Kafka

• Cause

  Si un mécanisme de transfert, tel qu'un proxy ou un mappage de port, existe entre Flink et Kafka, le client Kafka récupère l'adresse réseau interne du serveur Kafka plutôt que celle du proxy. Par conséquent, Flink parvient à se connecter au cluster Kafka mais ne peut ni consommer ni écrire de données, même si le chemin réseau est établi.

  Le processus de connexion entre le connecteur Flink Kafka et le serveur Kafka comporte deux étapes :

  1. Le client Kafka récupère les métadonnées auprès des brokers Kafka. Ces métadonnées incluent les adresses réseau de tous les brokers du cluster.

  2. Le connecteur Flink utilise ensuite ces adresses réseau pour consommer ou écrire des données.

• Dépannage

  Suivez ces étapes pour déterminer si un mécanisme de transfert, tel qu'un proxy ou un mappage de port, existe entre Flink et Kafka :

  1. Utilisez un outil en ligne de commande ZooKeeper (zkCli.sh ou zookeeper-shell.sh) pour vous connecter au cluster ZooKeeper utilisé par votre cluster Kafka.

  2. Exécutez la commande appropriée à votre cluster pour récupérer les métadonnées du broker Kafka.

     Vous pouvez généralement utiliser la commande get /brokers/ids/0 pour récupérer les métadonnées du broker Kafka. L'adresse de connexion se trouve dans le champ endpoints. Par exemple, connectez-vous via ZooKeeper Shell et exécutez get /brokers/ids/0 pour afficher les informations d'enregistrement du broker. Notez l'adresse configurée dans le champ endpoints du JSON retourné :

     # bin/zookeeper-shell.sh localhost:2181
     Connecting to localhost:2181
     Welcome to ZooKeeper!
     JLine support is disabled
     WATCHER::
     WatchedEvent state:SyncConnected type:None path:null
     get /brokers/ids/0
     {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://116.62.xxx:9092"],"jmx_port":-1,"host":"116.62.xxx","timestamp":"1614840078030","port":9092,"version":4}

  3. Utilisez une commande telle que ping ou telnet pour tester la connectivité depuis l'environnement Flink vers l'adresse indiquée dans le champ endpoints.

     Un échec de connexion indique qu'un mécanisme de transfert, tel qu'un proxy ou un mappage de port, existe entre Flink et Kafka.

• Solutions

  • N'utilisez pas de mécanisme de transfert. Établissez plutôt un chemin réseau direct entre Flink et Kafka. Cela permet à Flink de se connecter directement aux endpoints listés dans les métadonnées Kafka.

  • Contactez votre administrateur Kafka pour configurer l'adresse de transfert dans la propriété advertised.listeners sur les brokers Kafka. Ainsi, le client Kafka récupérera des métadonnées contenant l'adresse de transfert correcte.

    Remarque

    Seules les versions Kafka 0.10.2.0 et ultérieures prennent en charge l'ajout d'une adresse proxy aux listeners d'un broker Kafka.

  Pour plus d'informations sur ce fonctionnement, consultez KIP-103: Separate Internal and External traffic et Kafka client cannot connect to brokers.

Si la connectivité réseau entre Flink et Kafka est confirmée et que le problème persiste, vérifiez les causes non liées au réseau suivantes :

Vérification 1 : Stratégie d'offset de départ

Vérifiez le paramètre scan.startup.mode dans la clause WITH de la DDL de votre table source Kafka. Si la valeur est latest-offset, Flink lit uniquement les messages écrits après le démarrage du job. Si aucun nouveau message n'arrive après le lancement du job, celui-ci semble ne consommer aucune donnée.

Pour vérifier que de nouvelles données sont produites après le démarrage du job, utilisez un client consommateur Kafka afin de surveiller le topic en temps réel.

Vérification 2 : Incompatibilité de format de données

Assurez-vous que le paramètre format dans la clause WITH de votre table source Kafka correspond à l'encodage réel des messages dans le topic Kafka. Une incompatibilité de format entraîne des échecs de désérialisation, ce qui peut amener le job à ignorer silencieusement des messages ou à ne produire aucune sortie.

Pour inspecter le format réel des messages, utilisez un client consommateur Kafka afin de lire les octets bruts du topic et examiner la structure de la charge utile.

Absence de sortie de données des fenêtres temporelles événementielles Kafka

• Problème

  Un job ne produit aucune sortie lorsqu'il utilise une table source Kafka avec une fenêtre temporelle événementielle.

• Cause

  Une partition Kafka inactive peut empêcher la progression du watermark, ce qui bloque la production de sortie par la fenêtre temporelle événementielle.

• Solution

  1. Assurez-vous que toutes les partitions reçoivent des données.

  2. Pour activer la détection d'inactivité de la source, ajoutez le code suivant dans la section Other Configurations et enregistrez vos modifications. Pour des instructions détaillées, consultez Comment configurer les paramètres d'exécution personnalisés du job ?.

     table.exec.source.idle-timeout: 5

     Pour plus d'informations sur le paramètre table.exec.source.idle-timeout, consultez Configuration.

Validation des offsets dans Kafka

La validation d'un offset Kafka suit la position des données traitées, garantissant ainsi la cohérence et la fiabilité du traitement de flux en évitant la duplication ou la perte de données. Lorsqu'un checkpoint se termine avec succès, Flink valide l'offset de lecture correspondant dans Kafka. Si le mécanisme de checkpoint n'est pas activé, ou si l'intervalle entre les checkpoints est trop long, l'offset validé dans Kafka peut devenir obsolète, entraînant un retraitement ou une perte de données.

Analyser des JSON imbriqués avec le connecteur Kafka

{"data":[{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"},{"cola":"test1","colb":"test2"}]}

Se connecter à un cluster Kafka sécurisé

1. Dans la clause WITH de la DDL de votre table Kafka, ajoutez les configurations de sécurité pour l'authentification et le chiffrement. Pour une liste complète des options, consultez SECURITY.

   Important

   Préfixez tous les paramètres de configuration de sécurité avec properties.

   • Cet exemple montre comment configurer une table Kafka pour utiliser le mécanisme SASL PLAIN et fournir une configuration JAAS.

     CREATE TABLE KafkaTable (
       `user_id` BIGINT,
       `item_id` BIGINT,
       `behavior` STRING,
       `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
     ) WITH (
       'connector' = 'kafka',
       ...
       'properties.security.protocol' = 'SASL_PLAINTEXT',
       'properties.sasl.mechanism' = 'PLAIN',
       'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username=\"username\" password=\"password\";'
     );

   • Cet exemple montre comment utiliser le protocole de sécurité SASL_SSL avec le mécanisme SASL SCRAM-SHA-256.

     CREATE TABLE KafkaTable (
       `user_id` BIGINT,
       `item_id` BIGINT,
       `behavior` STRING,
       `ts` TIMESTAMP(3) METADATA FROM 'timestamp'
     ) WITH (
       'connector' = 'kafka',
       ...
       'properties.security.protocol' = 'SASL_SSL',
       /* SSL configuration */
       /* Path to the truststore (CA certificate) provided by the server */
       'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
       'properties.ssl.truststore.password' = 'test1234',
       /* If client-side authentication is required, configure the path to the keystore (private key) */
       'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
       'properties.ssl.keystore.password' = 'test1234',
       /* SASL configuration */
       /* Configure the SASL mechanism as SCRAM-SHA-256 */
       'properties.sasl.mechanism' = 'SCRAM-SHA-256',
       /* Configure JAAS */
       'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username=\"username\" password=\"password\";'
     );

     Remarque

     • Si properties.sasl.mechanism est SCRAM-SHA-256, utilisez org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule pour properties.sasl.jaas.config.

     • Si properties.sasl.mechanism est PLAIN, utilisez org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule pour properties.sasl.jaas.config.

2. Dans la section Additional dependency files de votre job, téléchargez tous les fichiers requis, tels que les certificats, les clés publiques et les clés privées.

   La plateforme stocke les fichiers téléchargés dans le répertoire /flink/usrlib. Pour les instructions de téléchargement, consultez déploiement de job.

   Important

   Si le mécanisme d'authentification sur votre broker Kafka est SASL_SSL mais que le mécanisme côté client est SASL_PLAINTEXT, le job échoue avec une exception OutOfMemory lors de la validation. Pour résoudre ce problème, assurez-vous que les mécanismes d'authentification côté client et côté serveur correspondent.

Conflits de noms de champs

• Problème

  Une source de données Kafka sérialise les messages en deux chaînes JSON distinctes : une pour la clé et une pour la valeur. Dans ce scénario, la clé et la valeur contiennent toutes deux un champ portant le même nom, comme le champ id dans l'exemple ci-dessous. L'analyse directe de ces données dans une table Flink provoque un conflit de noms de champs.

  • key

    {
       "id": 1
    }

  • value

    {
       "id": 100,
       "name": "flink"
    }

• Solution

  Utilisez la propriété key.fields-prefix pour éviter ce problème.

  CREATE TABLE kafka_table (
    -- Define the columns for the key and value fields
    key_id INT,
    value_id INT,
    name STRING
  ) WITH (
    'connector' = 'kafka',
    'topic' = 'test_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json',
    'json.ignore-parse-errors' = 'true',
    -- Specify the fields and data types for the key
    'key.format' = 'json',
    'key.fields' = 'id',
    'value.format' = 'json',
    'value.fields' = 'id, name',
    -- Add a prefix to fields from the key
    'key.fields-prefix' = 'key_'
  );

  Définir la propriété key.fields-prefix sur key_ indique au connecteur d'ajouter le préfixe key_ à tous les champs provenant de la clé du message. Par exemple, le champ id de la clé devient la colonne key_id dans la table Flink. Cela évite un conflit avec le champ id de la valeur, qui correspond à la colonne value_id.

  L'exécution de la requête SELECT * FROM kafka_table; produit le résultat suivant :

  key_id: 1,
  value_id: 100,
  name: flink

Dépanner une latence élevée d'une source Kafka

• Problème

  Lorsque vous lisez depuis une table source Kafka, la métrique currentEmitEventTimeLag affiche une valeur supérieure à 50 ans. Par exemple, plusieurs jobs Flink SQL sont à l'état running, mais la colonne business latency affiche une valeur anormalement élevée dépassant 19 160 jours, telle que 19160d 1h 59m 28s.

• Dépannage

  1. Tout d'abord, déterminez si le job est un job JAR ou un job SQL.

     Pour un job JAR, vérifiez que votre fichier pom.xml utilise la dépendance Kafka fournie par Realtime Compute for Apache Flink. La version open source du connecteur ne signale pas ces métriques.

  2. Vérifiez si toutes les partitions du topic Kafka en amont reçoivent des données en temps réel.

  3. Vérifiez si le timestamp dans les métadonnées du message Kafka est égal à 0 ou null.

     La latence d'une source Kafka est calculée en soustrayant l'horodatage du message de l'heure actuelle. Si un message n'a pas d'horodatage, la latence peut s'afficher comme étant supérieure à 50 ans. Vous pouvez vérifier l'horodatage de l'une des manières suivantes :

     • Pour les jobs SQL, vous pouvez récupérer l'horodatage du message en définissant une colonne de métadonnées. Pour plus d'informations, consultez Table source Kafka.

       CREATE TEMPORARY TABLE sk_flink_src_user_praise_rt (
           `timestamp` BIGINT ,
           `timestamp` TIMESTAMP METADATA,  --Metadata timestamp.
           ts as to_timestamp (
             from_unixtime (`timestamp`, 'yyyy-MM-dd HH:mm:ss')
           ),
           watermark for ts as ts - interval '5' second
         ) WITH (
           'connector' = 'kafka',
           'topic' = '',
           'properties.bootstrap.servers' = '',
           'properties.group.id' = '',
           'format' = 'json',
           'scan.startup.mode' = 'latest-offset',
           'json.fail-on-missing-field' = 'false',
           'json.ignore-parse-errors' = 'true'
         );

     • Écrivez un programme Java simple qui utilise le client KafkaConsumer pour lire un message et inspecter son horodatage.

Erreur : les tables 'upsert-kafka' nécessitent une PRIMARY KEY

• Problème

  ) WITH (
      'connector' = 'upsert-kafka',
      'topic' = 'flow_stay_duration',
      'properties.bootstrap.servers' = 'xxx',
      'key.format' = 'avro',
      'value.format' = 'avro'
  );
      insert into sink_ad_data_device_info
  org.apache.flink.table.api.ValidationException: SQL validation failed. Unable to create a sink for writing table 'vvp.default.sink_ad_data_device_info'.
  The cause is following: 'upsert-kafka' tables require to define a PRIMARY KEY constraint. The PRIMARY KEY specifies which columns should be read from or write to the Kafka message key. The PRIMARY KEY also defines records in the 'upsert-kafka' table should update or delete on which keys.
  Table options are:
  'connector'='upsert-kafka'
  'key.format'='avro'
  'properties.bootstrap.servers'='xxx'
  'topic'='flow_stay_duration'
  'value.format'='avro'
      at org.apache.flink.table.sqlserver.utils.FormatValidatorExceptionUtils.newValidationException(FormatValidatorExceptionUtils.java:41)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.formatException(ErrorConverter.java:123)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.toErrorDetail(ErrorConverter.java:60)
      at org.apache.flink.table.sqlserver.utils.ErrorConverter.toGrpcException(ErrorConverter.java:54)
      at org.apache.flink.table.sqlserver.FlinkSqlServiceImpl.validateAndGeneratePlan(FlinkSqlServiceImpl.java:979)
      at org.apache.flink.table.sqlserver.proto.FlinkSqlServiceGrpc$MethodHandlers.invoke(FlinkSqlServiceGrpc.java:3283)
      at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:172)
      at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:331)

• Cause

  Cette erreur survient lorsque la DDL ne définit pas de clé primaire. Utilisé comme table sink, le connecteur upsert-kafka consomme le flux de journal des modifications provenant de la logique en amont. Il écrit les données INSERT et UPDATE_AFTER dans Kafka. Pour les opérations DELETE, il envoie un message avec une valeur nulle afin de signaler la suppression du message associé à la clé correspondante. Flink se sert des colonnes de la clé primaire pour partitionner les données, garantissant ainsi l'ordre des messages partageant la même clé et leur routage vers la même partition pour les mises à jour ou suppressions.

• Solution

  Définissez une clé primaire dans la DDL.

Récupération d'un job Flink après fractionnement ou réduction d'un topic

Lorsqu'un topic DataHub lu par un job Flink subit un fractionnement ou une réduction, le thread temporisateur du connecteur DataHub détecte le changement du nombre de shards et provoque délibérément l'échec de la source. Le job déclenche alors un basculement.

Le job récupère automatiquement après le basculement, sans redémarrage manuel. Le processus de récupération se déroule comme suit :

• Le connecteur exécute à nouveau listShard et ajoute les nouveaux shards absents de l'état du checkpoint.

• La consommation des shards existants reprend aux offsets enregistrés dans le checkpoint.

• Les nouveaux shards commencent la consommation depuis l'offset le plus ancien (OLDEST).

Suppression de topics avec consommateurs actifs

La suppression ou la recréation d'un topic DataHub ayant des consommateurs actifs est impossible.

Paramètres endPoint et tunnelEndpoint

Échec de création de table DataHub avec NoPermissionException : dhs:ListShard

• Symptôme

  Lors du déploiement d'une table source ou sink DataHub par un job Flink, l'opération échoue avec une erreur similaire à la suivante :

  NoPermissionException: You have no permission to perform this action. Action: dhs:ListShard

• Cause

  Cette erreur provient d'une convention de nommage non standard des paramètres WITH dans la DDL DataHub. Le connecteur DataHub exige que les identifiants soient spécifiés sous la forme accessId et accessKey. L'utilisation des formes à points access.id et access.key empêche le connecteur de reconnaître les champs d'identification et de s'authentifier. Par conséquent, le connecteur tente de lister les shards sans identifiants valides, et DataHub renvoie une NoPermissionException.

  Bien que le message d'erreur mentionne un privilège dhs:ListShard manquant, la cause racine réside dans la non-reconnaissance des paramètres d'identification, et non dans une absence réelle d'autorisation IAM.

• Solution

  Dans votre DDL DataHub, renommez access.id en accessId et access.key en accessKey. Supprimez la table existante et recréez-la avec la clause WITH corrigée.

  Configuration incorrecte :

  CREATE TABLE datahub_source (...) WITH (
    'connector'       = 'datahub',
    'endPoint'        = 'https://dh-cn-hangzhou.aliyuncs.com',
    'project'         = 'your_project',
    'topic'           = 'your_topic',
    'access.id'       = 'your-access-key-id',      -- Incorrect: dotted form not recognized
    'access.key'      = 'your-access-key-secret'   -- Incorrect: dotted form not recognized
  );

  Configuration correcte :

  CREATE TABLE datahub_source (...) WITH (
    'connector'  = 'datahub',
    'endPoint'   = 'https://dh-cn-hangzhou.aliyuncs.com',
    'project'    = 'your_project',
    'topic'      = 'your_topic',
    'accessId'   = 'your-access-key-id',      -- Correct
    'accessKey'  = 'your-access-key-secret'   -- Correct
  );

  Pour la liste complète des paramètres WITH pris en charge, consultez la documentation du connecteur DataHub.

Lectures complètes et incrémentielles depuis des sources MaxCompute

Les sources MaxCompute effectuent des lectures complètes et incrémentielles via MaxCompute Tunnel. Le débit de lecture est limité par la bande passante de MaxCompute Tunnel.

Les tables sources MaxCompute peuvent-elles lire des données ajoutées ?

Non. Une fois le job Flink démarré, les nouvelles données ajoutées à une table source ou à une partition ne sont pas lues. Cela s'applique que la source soit en cours de lecture active ou ait terminé sa lecture. L'ajout de données de cette manière peut également provoquer un basculement du job.

Modification de la concurrence pour les jobs MaxCompute en pause

Pour les tables sources MaxCompute avec l'option useNewApi activée (activée par défaut), les jobs en mode streaming prennent en charge les modifications de concurrence après une pause et une reprise. Une table source MaxCompute lit séquentiellement les partitions correspondantes. Lors de la lecture d'une partition, elle répartit les données de cette partition entre les opérateurs parallèles. La modification de la concurrence n'affecte pas la distribution des données pour la partition en cours de traitement avant la pause. Le nouveau parallélisme prend effet uniquement lorsque le job commence à traiter la partition suivante. Ainsi, si un job traite une seule partition volumineuse, l'augmentation de la concurrence suivie de la reprise du job peut entraîner la lecture des données par seulement une partie des opérateurs MaxCompute.

Les modifications de concurrence ne sont prises en charge ni pour les jobs batch, ni pour les jobs où l'option useNewApi est définie sur false.

Pourquoi MaxCompute lit-il des partitions antérieures lorsque la position de départ est 2019-10-11 00:00:00 ?

Éviter les lectures incomplètes de données dans les nouvelles partitions

Erreur d'autorisation du connecteur MaxCompute

• Détails de l'erreur

  Pendant l'exécution du job, une erreur s'affiche sur la page de basculement ou dans le fichier TaskManager.log :

  ErrorMessage=Authorization Failed [4019], You have NO privilege'ODPS:***'

• Cause

  Les informations d'identité utilisateur spécifiées dans la définition DDL MaxCompute ne disposent pas des permissions requises pour accéder à MaxCompute.

• Solution

  Authentifiez-vous à l'aide d'un compte Alibaba Cloud, d'un utilisateur RAM ou d'un rôle RAM. Pour plus d'informations, consultez Authentification utilisateur.

Configuration du paramètre startPartition

Démarrage lent des jobs MaxCompute incrémentiels

Définition du paramètre de partition

Lecture depuis des partitions

• Lecture depuis des partitions statiques

  Pour lire des partitions statiques d'une table source ou d'une table de dimension, configurez le paramètre partition comme suit.

  Étape	Description	Exemple
  1	Pour une table de dimension, spécifiez chaque partition sous la forme 'partition_column_name=partition_value'. La valeur de partition doit être fixe. Pour une table source, spécifiez chaque partition sous la forme 'partition_column_name=partition_value'. La valeur de partition peut être fixe ou contenir un caractère générique (*). Ce caractère générique correspond à n'importe quelle chaîne, y compris une chaîne vide.	Pour lire les données de la colonne de partition dt avec la valeur 20220901, spécifiez dt=20220901. Pour lire les données des partitions de la colonne dt dont les valeurs commencent par 202209, spécifiez dt=202209* (applicable uniquement aux tables sources). Pour lire les données des partitions de la colonne dt dont les valeurs commencent par 2022 et se terminent par 01, spécifiez dt=2022*01 (applicable uniquement aux tables sources). Pour lire les données de toutes les partitions de la colonne dt, spécifiez dt=* (applicable uniquement aux tables sources).
  2	Triez les chaînes de partition de l'étape 1 par niveau de partition croissant, puis concaténez-les avec des virgules (sans espaces). La chaîne résultante constitue la valeur du paramètre partition. Vous pouvez spécifier uniquement les premiers niveaux de partition.	Une table possède une partition unique de premier niveau dt. Pour lire les données de la partition dt=20220901, spécifiez 'partition' = 'dt=20220901'. Une table comporte trois niveaux de partition : une partition de premier niveau dt, une partition de second niveau hh et une partition de troisième niveau mm. Pour lire les données de dt=20220901, hh=08 et mm=10, spécifiez 'partition' = 'dt=20220901,hh=08,mm=10'. Pour la même table, afin de lire les données de dt=20220901, hh=08 et n'importe quelle valeur pour mm, spécifiez 'partition' = 'dt=20220901,hh=08' or 'partition' = 'dt=20220901,hh=08,mm=*'. Pour la même table, afin de lire les données de dt=20220901, n'importe quelle valeur pour hh et mm=10, spécifiez 'partition' = 'dt=20220901,hh=*,mm=10'.

  Si ces étapes ne répondent pas à vos besoins de filtrage de partitions, ajoutez les conditions de filtre dans la clause WHERE de votre instruction SQL. Cela permet à l'optimiseur SQL d'utiliser le pushdown de partition pour le filtrage. Par exemple, pour lire des partitions d'une table à deux niveaux de partition (dt et hh) où dt est compris entre '20220901' et '20220903', et hh entre '09' et '17', utilisez une instruction SQL similaire à la suivante.

  CREATE TABLE maxcompute_table (
    content VARCHAR,
    dt VARCHAR,
    hh VARCHAR
  ) PARTITIONED BY (dt, hh) WITH ( 
     -- You must specify the partition columns with PARTITIONED BY to enable 
     -- partition pushdown in the SQL optimizer, which improves performance.
    'connector' = 'odps',
    ... -- Fill in required parameters such as accessId. You can omit the 'partition' parameter and let the SQL optimizer filter partitions.
  );
  SELECT content, dt, hh FROM maxcompute_table
  WHERE dt >= '20220901' AND dt <= '20220903' AND hh >= '09' AND hh <= '17'; -- Specify partition filters in the WHERE clause.

• Lecture de la partition ayant l'ordre lexicographique le plus élevé

  • Pour lire la partition lexicographiquement la plus grande d'une table source ou d'une table de dimension, définissez le paramètre partition sur 'max_pt()'.

  • Pour lire les deux partitions lexicographiquement les plus grandes d'une table source ou d'une table de dimension, définissez le paramètre partition sur 'max_two_pt()'.

  • Pour lire la partition lexicographiquement la plus grande qui possède également une partition .done correspondante dans une table source ou une table de dimension, définissez le paramètre partition sur 'max_pt_with_done()'.

  Généralement, la partition lexicographiquement la plus grande est la plus récemment créée. L'option max_pt_with_done() s'avère utile lorsque les données de la dernière partition pourraient ne pas être prêtes, et que vous souhaitez que la table de dimension lise temporairement une partition légèrement plus ancienne mais complète.

  Lorsque les données d'une partition sont prêtes, vous devez également créer une partition vide correspondante. Son nom correspond au nom de la partition de données suivi de .done. Par exemple, une fois les données de la partition dt=20220901 prêtes, créez une partition vide nommée dt=20220901.done. Lorsque vous définissez le paramètre partition sur max_pt_with_done(), la table de dimension lit uniquement les partitions ayant une partition .done correspondante. Les partitions de données sans partition .done sont temporairement ignorées. Pour plus d'informations, consultez Quelle est la différence entre max_pt() et max_pt_with_done() ?.

  Remarque

  Une table source détermine la partition ayant l'ordre lexicographique le plus élevé uniquement au démarrage du job. Elle s'arrête après avoir lu toutes les données et ne surveille pas les nouvelles partitions. Si vous devez lire continuellement de nouvelles partitions, utilisez le mode table source incrémentielle. Une table de dimension vérifie et lit les données les plus récentes à chaque actualisation.

Écriture dans des partitions

• Écriture dans des partitions statiques

  Pour écrire des données dans une partition statique d'une table de résultats, vous pouvez définir le paramètre partition en utilisant la même méthode que pour la lecture depuis des partitions statiques.

  Important

  Le paramètre partition d'une table de résultats ne prend pas en charge les caractères génériques (*).

• Écriture dans des partitions dynamiques

  Pour écrire dans des partitions dynamiques, où les valeurs de partition sont dérivées des données, définissez le paramètre de partition sur une liste de noms de colonnes de partition séparés par des virgules, triés par niveau de partition croissant. Par exemple, si une table comporte trois niveaux de partition, dt, hh et mm, spécifiez 'partition' = 'dt,hh,mm'.

Démarrage lent des jobs pour les tables sources MaxCompute

Choix d'un canal de données

Duplication de données dans les tables de résultats MaxCompute

La duplication de données dans une table de résultats MaxCompute écrite par un job Flink peut résulter des causes suivantes :

• Vérifiez la logique de votre job. Même si une contrainte de clé primaire est déclarée dans la table de résultats MaxCompute, Flink n'effectue pas de vérifications d'unicité lors de l'écriture dans un stockage externe. De plus, les tables non transactionnelles dans MaxCompute ne prennent pas en charge les contraintes de clé primaire. Par conséquent, si la logique de votre job Flink génère des données dupliquées, ces doublons seront écrits dans la table MaxCompute.

• Vérifiez si plusieurs jobs Flink écrivent simultanément dans la même table MaxCompute. Comme mentionné précédemment, MaxCompute n'applique pas les contraintes de clé primaire. Si plusieurs jobs Flink produisent les mêmes résultats, ils créeront des enregistrements dupliqués dans la table.

• Le job Flink échoue pendant un checkpoint lors de l'utilisation de Batch Tunnel. Lorsqu'un échec survient pendant un checkpoint, les données de la table de résultats peuvent avoir déjà été validées sur le serveur. Par conséquent, lorsque le job récupère depuis le dernier checkpoint réussi, il peut écrire des données dupliquées pour la période comprise entre le dernier checkpoint réussi et l'échec.

• Un basculement de job Flink survient lors de l'utilisation de Stream Tunnel. Lors de l'écriture dans MaxCompute avec Stream Tunnel, les données sont validées sur le serveur MaxCompute entre les checkpoints. Si le job bascule et récupère depuis le dernier checkpoint, il peut écrire des données dupliquées qui ont été traitées après la fin du checkpoint mais avant le basculement. Pour plus d'informations, consultez Choix d'un canal de données. Pour éviter ce type de duplication, vous pouvez passer en mode Batch Tunnel.

• Un job Flink utilisant Batch Tunnel bascule ou redémarre après avoir été annulé (par exemple, déclenché par Autopilot). Dans les versions antérieures à vvr-6.0.7-flink-1.15, le job valide les données dans la table de résultats MaxCompute lors de l'arrêt. Par conséquent, lorsque le job Flink s'arrête puis récupère depuis le dernier checkpoint, il peut créer des données dupliquées pour la période comprise entre le dernier checkpoint et l'arrêt. Pour résoudre ce problème, mettez à jour votre version Flink vers vvr-6.0.7-flink-1.15 ou ultérieure.

Échec du job MaxCompute avec 'Invalid partition spec'

• Cause : Cette erreur survient lorsque les données écrites dans MaxCompute contiennent des valeurs invalides dans la colonne de partition. Les valeurs invalides incluent les chaînes vides, les valeurs nulles ou les valeurs contenant un signe égal (=), une virgule (,) ou une barre oblique (/).

• Solution : Vérifiez que les valeurs dans la colonne de partition de vos données sources sont valides.

Erreur 'No more available blockId' dans les jobs MaxCompute

• Cause : Le nombre de blocs écrits dans la table de résultats MaxCompute a dépassé la limite. Cela est généralement causé par la vidange trop fréquente de petites quantités de données.

• Solution : Ajustez les paramètres batchSize et flushIntervalMs.

Utilisation du hint SHUFFLE_HASH

-- Create a source table and three dimension tables.
CREATE TABLE source_table (k VARCHAR, v VARCHAR) WITH ( ... );
CREATE TABLE dim_1 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
CREATE TABLE dim_2 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
CREATE TABLE dim_3 (k VARCHAR, v VARCHAR) WITH ('connector' = 'odps', 'cache' = 'ALL', ... );
-- Specify the names of the dimension tables to distribute in the SHUFFLE_HASH hint.
SELECT /*+ SHUFFLE_HASH(dim_1), SHUFFLE_HASH(dim_3) */
k, s.v, d1.v, d2.v, d3.v
FROM source_table AS s
INNER JOIN dim_1 FOR SYSTEM_TIME AS OF PROCTIME() AS d1 ON s.k = d1.k
LEFT JOIN dim_2 FOR SYSTEM_TIME AS OF PROCTIME() AS d2 ON s.k = d2.k
LEFT JOIN dim_3 FOR SYSTEM_TIME AS OF PROCTIME() AS d3 ON s.k = d3.k;

Configuration de CacheReloadTimeBlackList

Erreur : java.io.EOFException: SSL peer shut down incorrectly

• Détails de l'erreur

  Caused by: java.io.EOFException: SSL peer shut down incorrectly
      at sun.security.ssl.SSLSocketInputRecord.decodeInputRecord(SSLSocketInputRecord.java:239) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketInputRecord.decode(SSLSocketInputRecord.java:190) ~[?:1.8.0_302]
      at sun.security.ssl.SSLTransport.decode(SSLTransport.java:109) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1392) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1300) ~[?:1.8.0_302]
      at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:435) ~[?:1.8.0_302]
      at com.mysql.cj.protocol.ExportControlled.performTlsHandshake(ExportControlled.java:347) ~[?:?]
      at com.mysql.cj.protocol.StandardSocketFactory.performTlsHandshake(StandardSocketFactory.java:194) ~[?:?]
      at com.mysql.cj.protocol.a.NativeSocketConnection.performTlsHandshake(NativeSocketConnection.java:101) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.negotiateSSLConnection(NativeProtocol.java:308) ~[?:?]
      at com.mysql.cj.protocol.a.NativeAuthenticationProvider.connect(NativeAuthenticationProvider.java:204) ~[?:?]
      at com.mysql.cj.protocol.a.NativeProtocol.connect(NativeProtocol.java:1369) ~[?:?]
      at com.mysql.cj.NativeSession.connect(NativeSession.java:133) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.connectOneTryOnly(ConnectionImpl.java:949) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:819) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:449) ~[?:?]
      at com.mysql.cj.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:242) ~[?:?]
      at com.mysql.cj.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:198) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider.getOrEstablishConnection(SimpleJdbcConnectionProvider.java:128) ~[?:?]
      at org.apache.flink.connector.jdbc.internal.AbstractJdbcOutputFormat.open(AbstractJdbcOutputFormat.java:54) ~[?:?]
      ... 14 more

• Cause

  Cette erreur survient généralement lorsqu'une base de données MySQL a le protocole SSL activé, mais que la connexion SSL du client n'est pas configurée correctement. Par exemple, avec le pilote MySQL 8.0.27 et une base de données MySQL avec SSL activé, cette erreur se produit car la méthode d'accès par défaut du pilote n'utilise pas SSL.

• Solution

  Ajoutez characterEncoding=utf-8&useSSL=false au paramètre URL de la table de dimension MySQL. Par exemple :

  'url'='jdbc:mysql://***.***.***.***:3306/test?characterEncoding=utf-8&useSSL=false'

Modification du type de clé MySQL** bigint unsigned**

Flink ne prend pas en charge le type de données bigint unsigned. Afin d'éviter tout risque de dépassement de capacité, Flink mappe la clé primaire bigint unsigned vers le type decimal. Lors de la synchronisation vers Hologres, le système convertit cette colonne en type text, car Hologres ne supporte ni le type bigint unsigned, ni l'utilisation du type decimal comme clé primaire.

Prenez ce comportement en compte lors de vos phases de conception et de développement. Pour conserver le type decimal sur cette colonne, créez manuellement la table dans Hologres avant de lancer la synchronisation. Dans cette table, vous avez la possibilité de définir une autre colonne comme clé primaire ou de ne spécifier aucune clé primaire. Toutefois, cette approche risque d'entraîner des doublons, puisque la clé primaire d'origine n'impose plus l'unicité. Vous devrez alors gérer cette problématique au niveau applicatif, par exemple en tolérant les doublons ou en mettant en œuvre une logique de déduplication.

De Flink vers RDS : mise à jour ou insertion

Si une clé primaire est définie dans le DDL, le connecteur utilise une instruction INSERT INTO tablename(field1,field2, field3, ...) VALUES(value1, value2, value3, ...) ON DUPLICATE KEY UPDATE field1=value1,field2=value2, field3=value3, ...;. Cette instruction insère un nouvel enregistrement si la clé primaire n'existe pas, ou met à jour l'enregistrement existant dans le cas contraire. Si aucune clé primaire n'est déclarée dans le DDL, le connecteur insère les nouveaux enregistrements via une instruction insert into.

Utilisation d'un index unique avec GROUP BY

• Vous devez déclarer l'index unique dans la clause GROUP BY de votre job.

• Si une table RDS utilise une clé primaire auto-incrémentée, ne la déclarez pas en tant que PRIMARY KEY dans le job Flink.

Mappage INT UNSIGNED : de MySQL vers Flink SQL

Le pilote JDBC MySQL mappe les entiers non signés vers des types de données Java plus larges afin de préserver la précision. Plus précisément, le pilote mappe les valeurs INT UNSIGNED de MySQL vers le type LONG de Java, que Flink SQL traite ensuite comme un BIGINT. De même, il mappe les valeurs BIGINT UNSIGNED de MySQL vers le type BigInteger de Java, que Flink SQL traite comme un DECIMAL(20, 0).

Erreur : valeur de chaîne incorrecte

• Détails de l'erreur

  Caused by: java.sql.BatchUpdateException: Incorrect string value: '\xF0\x9F\x98\x80\xF0\x9F...' for column 'test' at row 1
  at sun.reflect.GeneratedConstructorAccessor59.newInstance(Unknown Source)
  at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
  at com.mysql.cj.util.Util.getInstance(Util.java:167)
  at com.mysql.cj.util.Util.getInstance(Util.java:174)
  at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchedInserts(ClientPreparedStatement.java:755)
  at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:426)
  at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:796)
  at com.alibaba.druid.pool.DruidPooledPreparedStatement.executeBatch(DruidPooledPreparedStatement.java:565)
  at com.alibaba.ververica.connectors.rds.sink.RdsOutputFormat.executeSql(RdsOutputFormat.java:488)
  ... 15 more

• Cause

  Les données contiennent des caractères spéciaux ou utilisent un encodage de caractères non pris en charge par la base de données.

• Solution

  Ajoutez characterEncoding=UTF-8 à l'URL lors de la connexion à une base de données MySQL via JDBC. Par exemple : jdbc:mysql://<internal address>/<databaseName>?characterEncoding=UTF-8.

Interblocages dans MySQL (TDDL/RDS)

• Problème

  Un interblocage survient lors de l'écriture de données dans MySQL (TDDL/RDS).

  Important

  Dans Realtime Compute for Apache Flink, si vous utilisez une base de données relationnelle telle que MySQL comme destination (via le connecteur TDDL/RDS), des écritures fréquentes sur une table ou une ressource peuvent provoquer des interblocages.

  Supposons qu'une opération INSERT nécessite l'acquisition séquentielle de deux verrous, (A,B). Le verrou A est un verrou d'intervalle. Deux transactions sont en cours, (T1,T2), et le schéma de la table est (id(auto-incrementing primary key),nid(unique key)). T1 contient deux instructions, insert(null,2),(null,1), tandis que T2 n'en contient qu'une seule, insert(null,2).

  1. À l'instant t, T1 exécute sa première instruction INSERT. T1 détient alors les deux verrous (A,B).

  2. À l'instant t+1, T2 démarre une opération d'insertion et doit attendre le verrou A pour verrouiller l'intervalle (-inf,2]. À ce moment-là, le verrou A est détenu par T1 et couvre déjà l'intervalle (-inf,2]. En raison de cette relation d'inclusion, T2 dépend de la libération du verrou A par T1.

  3. À l'instant t+2, T1 exécute sa seconde instruction INSERT, qui requiert le verrou A sur l'intervalle (-inf,1]. Cet intervalle étant un sous-ensemble de (-inf,2], T1 doit se mettre en file d'attente et attendre que T2 libère le verrou. Par conséquent, T1 devient dépendant de T2 pour la libération du verrou A.

  Un interblocage se produit car T1 et T2 attendent désormais mutuellement la libération de leurs verrous respectifs.

• RDS/TDDL et Tablestore utilisent des mécanismes de verrouillage différents.

  • RDS/TDDL : Le verrou de ligne d'InnoDB s'applique à l'index, et non à un enregistrement individuel. Par conséquent, même lors de l'accès à des lignes différentes, un conflit de verrouillage peut survenir si ces lignes partagent la même clé d'index. Cela peut bloquer les mises à jour sur toute une plage de données.

  • Tablestore : Utilise un verrou de ligne unique, qui n'affecte pas les mises à jour d'autres données.

• Solutions aux interblocages

  Pour les scénarios d'écriture à QPS/TPS élevé ou à forte concurrence, utilisez Tablestore comme table de résultats afin d'éviter les interblocages. De manière générale, l'utilisation de TDDL ou RDS comme table de résultats pour un job Flink n'est pas recommandée.

  Si vous devez impérativement utiliser une base de données relationnelle comme MySQL comme nœud de destination, tenez compte des recommandations suivantes :

  • Assurez-vous qu'aucune autre charge de travail ne lit ou n'écrit simultanément dans les mêmes tables.

  • Si le volume de données du job est faible, essayez d'écrire les données en mode monothread. Notez toutefois que dans des contextes de QPS/TPS élevé et de forte concurrence, cette approche réduit les performances d'écriture.

  • Évitez autant que possible d'utiliser une clé unique, car les écritures dans des tables possédant une clé unique peuvent provoquer des interblocages. Si vos exigences métier imposent une clé unique, définissez-la en classant les colonnes de la plus sélective à la moins sélective. Cela réduit considérablement la probabilité d'interblocage. Par exemple, placez une colonne contenant un hachage MD5 avant la colonne day_time(20171010) afin de garantir que la clé unique soit définie avec les colonnes les plus sélectives en premier.

  • Utilisez le sharding de base de données et le partitionnement de tables en fonction des caractéristiques de votre charge de travail pour répartir les écritures sur plusieurs tables. Pour plus de détails sur la mise en œuvre, contactez votre administrateur de base de données.

Échec de la mise à jour de la structure de la table en aval

La synchronisation de la structure des tables ne suit pas les instructions DDL. Elle détecte plutôt les changements de schéma en comparant des enregistrements consécutifs. Si une modification DDL survient sans changement ultérieur des données en amont, la structure de la table en aval ne sera pas mise à jour. Pour plus de détails, consultez la rubrique Politique de synchronisation des modifications de structure de table.

Erreur de délai d'expiration de réponse de fin de split

Cette erreur survient lorsqu'une utilisation élevée du CPU par une tâche l'empêche de répondre aux requêtes RPC du coordinateur. Pour résoudre ce problème, augmentez les ressources CPU allouées au TaskManager sur la page de configuration des ressources.

Impact des changements de schéma pendant le chargement complet

Une modification de schéma durant la phase de chargement complet peut entraîner l'échec du job ou empêcher la synchronisation du changement de schéma. Pour y remédier, arrêtez le job, supprimez la table en aval, puis redémarrez le job sans état.

Changements de schéma non pris en charge lors de la synchronisation CTAS/CDAS

Resynchronisez les données de la table. Pour ce faire, arrêtez le job, supprimez la table en aval, puis redémarrez le job de synchronisation sans état initial. Évitez d'effectuer de telles modifications incompatibles, car le job échouera à nouveau lors du redémarrage. Pour plus d'informations sur les changements de schéma pris en charge, consultez la rubrique Instruction CREATE TABLE AS (CTAS).

Mises à jour par rétractation dans ClickHouse

Les mises à jour par rétractation sont prises en charge pour une table de résultats ClickHouse si vous spécifiez une clé primaire dans le DDL de la table de résultats Flink et si vous définissez le paramètre ignoreDelete sur false. Cependant, cela entraîne une baisse significative des performances.

ClickHouse est un système de gestion de base de données orienté colonne conçu pour le traitement analytique en ligne (OLAP), et sa prise en charge des opérations UPDATE et DELETE est limitée. Si vous spécifiez une clé primaire dans le DDL Flink, le connecteur tente d'utiliser ALTER TABLE UPDATE et ALTER TABLE DELETE pour mettre à jour et supprimer des données. Ces opérations sont très inefficaces.

Visibilité des données dans ClickHouse

• Pour une table de résultats ClickHouse avec la sémantique exactly-once semantics désactivée (comportement par défaut), les données deviennent visibles dès que le tampon est vidé. Le système vide automatiquement ce tampon lorsque le nombre d'enregistrements atteint la valeur batchSize ou lorsque le temps écoulé depuis la dernière écriture dépasse flushIntervalMs. Il n'est pas nécessaire d'attendre la fin d'un checkpoint.

• Pour une table de résultats ClickHouse avec la sémantique exactly-once semantics activée, les données ne deviennent visibles qu'après la réussite du checkpoint correspondant.

Consultation des résultats d'impression

La jointure de table de dimension ne retourne aucune donnée

Vérifiez que le schéma — y compris les types de données et les noms de colonnes — défini dans l'instruction DDL correspond exactement à celui de la table physique.

max_pt() et max_pt_with_done()

Erreur de job d'écriture Paimon : « Heartbeat of TaskManager timed out »

La cause la plus probable de cette erreur est une mémoire heap insuffisante sur le TaskManager. Paimon utilise principalement la mémoire heap des manières suivantes :

• Chaque instance parallèle de l'opérateur d'écriture pour une table à clé primaire Paimon dispose d'un tampon mémoire dédié au tri. La taille de ce tampon est contrôlée par la propriété de table write-buffer-size, dont la valeur par défaut est de 256 Mo.

• Paimon utilise le format de fichier ORC par défaut, ce qui nécessite un tampon mémoire supplémentaire pour convertir les données en mémoire en format colonnaire par lots. La taille de ce tampon est contrôlée par la propriété de table orc.write.batch-size, dont la valeur par défaut est 1024, ce qui signifie que le tampon contient 1024 lignes de données.

• Chaque bucket modifié possède un objet writer dédié pour écrire ses données.

Compte tenu de ces modes d'utilisation, voici les causes potentielles d'une mémoire heap insuffisante ainsi que leurs solutions :

• La valeur de write-buffer-size est trop élevée.

  Essayez de réduire ce paramètre. Toutefois, un tampon trop petit peut entraîner des écritures fréquentes sur disque et déclencher plus souvent des compactages de petits fichiers, ce qui affecte les performances d'écriture.

• Un enregistrement de données individuel est trop volumineux.

  Par exemple, si un enregistrement contient un champ JSON de 4 Mo, le tampon ORC peut atteindre 4 Mo × 1024 = 4 Go, consommant ainsi une quantité importante de mémoire heap. Deux solutions s'offrent à vous :

  • Réduisez la valeur de orc.write.batch-size.

  • Si vous n'avez pas besoin d'effectuer des requêtes ad hoc (OLAP) sur la table de résultats Paimon et que vous nécessitez uniquement une consommation par lots ou en streaming, vous pouvez définir les propriétés de table 'file.format' = 'avro' et 'metadata.stats-mode' = 'none' lors de la création de la table. Cela fait passer la table au format Avro et désactive la collecte de statistiques.

    Remarque

    Ces paramètres ne peuvent être définis que lors de la création de la table. Ils ne peuvent pas être modifiés ultérieurement via une instruction ALTER TABLE ou un hint SQL après la création de la table.

• L'écriture simultanée dans un trop grand nombre de partitions ou la présence d'un nombre excessif de buckets par partition crée un nombre trop important d'objets writer.

  Vérifiez la configuration de vos colonnes de partition pour vous assurer qu'elle est appropriée. Assurez-vous qu'une requête SQL incorrecte ne provoque pas l'écriture inattendue de données dans la colonne de partition. Vérifiez également que le nombre de buckets est raisonnable. Comme bonne pratique, la taille totale des données par bucket devrait avoisiner les 2 Go et ne pas dépasser 5 Go. Pour plus de détails sur l'ajustement du nombre de buckets, consultez la rubrique Ajuster le nombre de buckets pour une table à buckets fixes.

Erreur : « Sink materializer must not be used with Paimon sink »

L'opérateur sink materializer gère les données désordonnées provenant de jointures en cascade dans les jobs en streaming. Cependant, dans les jobs écrivant dans une table Paimon, cet opérateur introduit une surcharge et peut produire des résultats incorrects lorsqu'une agrégation est utilisée. N'utilisez pas l'opérateur sink materializer avec un sink Paimon.

Vous pouvez désactiver l'opérateur sink materializer en définissant le paramètre table.exec.sink.upsert-materialize sur false à l'aide d'une instruction SET ou en tant que paramètre d'exécution. Si vous devez également gérer des données désordonnées, consultez la rubrique Gestion des données désordonnées.

Paimon : Erreur File deletion conflicts detected ou LSM conflicts detected

Cette erreur peut survenir pour les raisons suivantes :

• Plusieurs jobs écrivent simultanément dans la même partition d'une même table Paimon. Dans ce cas, Paimon résout le conflit par un mécanisme de failover and restart. Ce comportement est attendu ; aucune action n'est requise si l'erreur ne se reproduit pas.

• Le job est restauré à partir d'un état obsolète, ce qui provoque la récurrence de l'erreur. Pour résoudre ce problème, restaurez le job à partir de son état le plus récent ou start it without state.

• Paimon ne prend pas en charge les écritures séparées issues de plusieurs instructions INSERT au sein d'un même job. Utilisez plutôt une instruction UNION ALL pour écrire plusieurs flux de données dans la table Paimon.

• La concurrence du nœud Global Committer ou du nœud Compaction Coordinator (lors de l'écriture dans une table Append Scalable) est supérieure à 1. La concurrency de ces nœuds doit être fixée à 1 pour garantir la data consistency.

Erreur « File xxx not found » dans un job de consommation Paimon

La consommation des tables Paimon repose sur des fichiers snapshot. Si la période de rétention des snapshots est trop courte ou si le job de consommation manque d'efficacité, les fichiers snapshot peuvent expirer et être supprimés avant la fin du job. Cela provoque l'échec du job de consommation.

Pour résoudre ce problème, vous pouvez ajuster la période de rétention des fichiers snapshot, spécifier un ID de consommateur, ou optimiser le job de consommation. Pour vérifier quels fichiers snapshot sont disponibles et connaître leurs horodatages de création, consultez la table système Snapshots.

Erreur de job Paimon : No space left on device

• Un nombre excessif de fichiers de cache peut provoquer cette erreur si votre job exécute une requête Paimon, par exemple en utilisant une table Paimon comme table de dimension ou en définissant changelog-producer='lookup'. Pour éviter cela, utilisez des Hints SQL afin de configurer les paramètres suivants, qui limitent l'espace disque maximal et la durée de rétention du cache de requête.

  • lookup.cache-max-disk-size : Espace disque local maximal que le cache de requête peut utiliser. Les valeurs recommandées sont 256 Mo, 512 Mo et 1 Go.

  • lookup.cache-file-retention : Durée de rétention des fichiers du cache de requête. Les valeurs recommandées sont 30 min, 15 min, ou un intervalle plus court.

• Pour les jobs écrivant dans une table Paimon, utilisez des Hints SQL pour définir les paramètres suivants. Ces réglages limitent la taille des fichiers temporaires locaux pendant le processus d'écriture, prévenant ainsi les pénuries d'espace disque.

  • write-buffer-spillable : Contrôle si le tampon d'écriture peut déborder sur le disque. Définir ce paramètre sur false empêche totalement le tampon d'utiliser de l'espace disque.

  • write-buffer-spill.max-disk-size : Espace disque maximal que le tampon d'écriture peut utiliser lors du débordement. Les valeurs recommandées sont 256 Mo, 512 Mo et 1 Go.

Gestion des fichiers Paimon sur OSS

• Paimon conserve les fichiers de données historiques pour permettre l'accès aux versions précédentes d'une table. Vous pouvez ajuster la politique de rétention de ces fichiers pour gérer le stockage. Pour des instructions détaillées, consultez la rubrique Nettoyer les données expirées.

• Une configuration inappropriée des colonnes de partition ou un nombre excessif de buckets peut également causer ce problème. Comme bonne pratique, visez une taille de données d'environ 2 Go par bucket, avec un maximum de 5 Go. Pour plus d'informations, consultez la rubrique Partitionnement en buckets.

• Par défaut, les fichiers de données sont enregistrés au format ORC. Pour réduire la taille totale des fichiers de données, vous pouvez utiliser le format de compression ZSTD en définissant le paramètre de table 'file.compression' = 'zstd' lors de la création de la table.

  Remarque

  Ce paramètre ne peut être défini que lors de la création de la table et ne peut pas être modifié ultérieurement via une instruction ALTER TABLE ou un hint SQL.

La visibilité des données dépend de l'intervalle de checkpoint****

Oui. Paimon s'appuie sur les checkpoints pour garantir la sémantique exactly-once. Les données sont validées (committed) et deviennent visibles en aval uniquement à la fin du checkpoint. Avant cette validation, les données présentes dans le tampon local sont vidées vers le système de fichiers distant, mais elles ne sont pas encore lisibles.

Augmentation lente de la mémoire dans les jobs Paimon de longue durée

• Une augmentation de l'utilisation de la mémoire est normale si le débit (rps) du job augmente lui aussi progressivement.

• Si vous utilisez un catalogue de système de fichiers Paimon pour lire ou écrire sur OSS, assurez-vous de configurer les paramètres de catalogue fs.oss.endpoint, fs.oss.accessKeyId et fs.oss.accessKeySecret. Dans le cas contraire, le job Flink peut subir une fuite de mémoire lente, un problème connu au sein de la communauté.

Décalage de fuseau horaire de 8 heures lorsqu'un job Flink SQL consomme des données Paimon et écrit dans Kafka

Symptôme

Un job Flink SQL consomme une table Paimon et écrit dans Kafka. Même lorsque le champ temporel de la table source est de type STRING, un décalage de 8 heures apparaît dans les données Kafka en aval. La définition de paramètres au niveau SQL tels que table.local-time-zone n'a aucun effet.

Cause

Paimon utilise le fuseau horaire système de la JVM du TaskManager lors de la lecture et du filtrage des données. Lorsque le fuseau horaire par défaut du TaskManager (généralement Asia/Shanghai, UTC+8) ne correspond pas au fuseau horaire des données sources, un décalage de 8 heures est introduit. Le paramètre table.local-time-zone affecte uniquement la sémantique temporelle de la couche Flink SQL et ne peut pas modifier le fuseau horaire système de la JVM du TaskManager.

Solution

Dans la configuration Flink, sous les Runtime parameters du job, ajoutez le paramètre suivant pour définir le fuseau horaire système de la JVM du TaskManager sur UTC :

env.java.opts.taskmanager: -Duser.timezone=UTC

Enregistrez la configuration et redémarrez le job.

IllegalArgumentException : timeout value is negative

• Détails de l'erreur

  2021-02-24 15:14:58
  java.lang.RuntimeException: java.lang.RuntimeException: java.lang.IllegalArgumentException: timeout value is negative
      at com.alibaba.ververica.connectors.common.source.reader.ParallelReader.run(ParallelReader.java:166)
      at com.alibaba.ververica.connectors.common.source.AbstractParallelSourceBase.run(AbstractParallelSourceBase.java:205)
      at com.alibaba.ververica.connectors.metaq.source.MetaQRowDataSource.run(MetaQRowDataSource.java:84)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
      at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
      at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213)
  Caused by: java.lang.RuntimeException: java.lang.IllegalArgumentException: timeout value is negative
      at com.alibaba.ververica.connectors.common.source.reader.ParallelReader.runImpl(ParallelReader.java:244)

• Cause de l'erreur

  Lorsqu'aucun nouveau message MQ n'est consommé, le thread MetaQSource se met en veille pendant un intervalle défini par le paramètre pullIntervalMs, dont la valeur par défaut est -1. Le job échoue alors avec une exception IllegalArgumentException car une durée de mise en veille ne peut pas être négative.

• Solution

  Définissez le paramètre pullIntervalMs sur une valeur non négative.

Détection des changements de partition

• Pour les versions de Realtime Compute for Apache Flink antérieures à la version 6.0.2, l'opérateur source récupère le nombre de partitions toutes les 5 à 10 minutes. Un basculement (failover) est déclenché si le nombre de partitions diffère lors de trois vérifications consécutives. Par conséquent, la source initie un basculement dans un délai de 10 à 30 minutes. Après le redémarrage du job, celui-ci lit les données à partir de l'ensemble mis à jour des partitions.

• Pour les versions 6.0.2 et ultérieures de Realtime Compute for Apache Flink, l'opérateur source récupère le nombre de partitions toutes les 5 minutes par défaut. Lorsque de nouvelles partitions sont découvertes, elles sont assignées directement à l'opérateur source sur le TaskManager, qui commence alors à lire les données. Aucun basculement de job n'est nécessaire, ce qui permet à la source de détecter les changements de partition dans un délai de 1 à 5 minutes.

Erreur : Backpressure exceeds reject limit

• Détails de l'erreur

  26      at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:47)
  27      at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:911)
  28      at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:879)
  29      ... 13 more
  30  Caused by: java.lang.RuntimeException: Rpc Exception failed errorCount=6 with RpcException: request niagara.table.proto.UpsertRecordBatchRequest@b53e2d74 failed on final try 4, maxAttempts=4, sn=11.117.xxx, errorCode=11, msg=BackPresure Exceed Reject Limit [method:UpsertRecordBatch,transaction_id:xxx,table_id:xxx,table_version:128, actor_id:74538 xxx,worker_address:11.117.xxx]
  31          at com.alibaba.ververica.connectors.hologres.sink.HologresOutputFormat.sync(HologresOutputFormat.java:264)
  32          at com.alibaba.ververica.connectors.common.sink.OutputFormatSinkFunction.snapshotState(OutputFormatSinkFunction.java:91)
  33          at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.trySnapshotFunctionState(StreamingFunctionUtils.java:128)
  34          at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.snapshotFunctionState(StreamingFunctionUtils.java:101)
  35          at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotState(AbstractUdfStreamOperator.java:90)
  36          at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:186)
  37          ... 23 more

• Cause

  La pression d'écriture sur l'instance Hologres est trop élevée.

• Solution

  Contactez le support technique Hologres en fournissant les informations de votre instance pour demander une montée en charge.

Erreur : remaining connection slots are reserved for non-replication superuser connections

• Détails de l'erreur

  Caused by: com.alibaba.hologres.client.exception.HoloClientWithDetailsException: failed records 1, first:Record{schema=org.postgresql.model.TableSchema@188365, values=[f06b41455c694d24a18d0552b8b0****, com.chot.tpfymnq.meta, 2022-04-02 19:46:40.0, 28, 1, null], bitSet={0, 1, 2, 3, 4}},first err:[106]FATAL: remaining connection slots are reserved for non-replication superuser connections
      at com.alibaba.hologres.client.impl.Worker.handlePutAction(Worker.java:406) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.run(Worker.java:118) ~[?:?]
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_302]
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_302]
      ... 1 more
  Caused by: com.alibaba.hologres.org.postgresql.util.PSQLException: FATAL: remaining connection slots are reserved for non-replication superuser connections
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2553) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.readStartupMessages(QueryExecutorImpl.java:2665) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.QueryExecutorImpl.<init>(QueryExecutorImpl.java:147) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.v3.ConnectionFactoryImpl.openConnectionImpl(ConnectionFactoryImpl.java:273) ~[?:?]
      at com.alibaba.hologres.org.postgresql.core.ConnectionFactory.openConnection(ConnectionFactory.java:51) ~[?:?]
      at com.alibaba.hologres.org.postgresql.jdbc.PgConnection.<init>(PgConnection.java:240) ~[?:?]
      at com.alibaba.hologres.org.postgresql.Driver.makeConnection(Driver.java:478) ~[?:?]
      at com.alibaba.hologres.org.postgresql.Driver.connect(Driver.java:277) ~[?:?]
      at java.sql.DriverManager.getConnection(DriverManager.java:674) ~[?:1.8.0_302]
      at java.sql.DriverManager.getConnection(DriverManager.java:217) ~[?:1.8.0_302]
      at com.alibaba.hologres.client.impl.ConnectionHolder.buildConnection(ConnectionHolder.java:122) ~[?:?]
      at com.alibaba.hologres.client.impl.ConnectionHolder.retryExecute(ConnectionHolder.java:195) ~[?:?]
      at com.alibaba.hologres.client.impl.ConnectionHolder.retryExecute(ConnectionHolder.java:184) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.doHandlePutAction(Worker.java:460) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.handlePutAction(Worker.java:389) ~[?:?]
      at com.alibaba.hologres.client.impl.Worker.run(Worker.java:118) ~[?:?]
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_302]
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_302]
      ... 1 more

• Cause

  La limite de connexions de l'instance Hologres a été dépassée.

• Solution

  • Vérifiez le paramètre app_name des connexions à chaque Frontend (FE) pour comptabiliser les connexions client Hologres provenant du flink-connector.

  • Vérifiez si d'autres jobs sont connectés à Hologres.

  • Libérez des connexions. Pour plus d'informations, consultez la rubrique Gestion des connexions.

No table is defined in publication

• Détails de l'erreur

  La suppression puis la recréation d'une table portant le même nom peut amener un job à signaler l'erreur no table is defined in publication.

• Cause

  La suppression d'une table n'efface pas la publication qui lui est associée.

• Solution

  1. Dans Hologres, exécutez la commande select * from pg_publication where pubname not in (select pubname from pg_publication_tables); pour rechercher les informations de publication qui n'ont pas été nettoyées lors de la suppression d'une table.

  2. Exécutez l'instruction drop publication xx; pour supprimer la publication restante.

  3. Redémarrez le job.

Intervalle de checkpoint et visibilité des données

L'intervalle de checkpoint du connecteur sink Flink Hologres ne contrôle pas directement la visibilité des données dans Hologres. Son rôle principal est de définir le SLA pour la reprise après incident.

Le connecteur Hologres ne prend pas en charge les transactions. Il vide périodiquement le tampon mémoire vers la base de données. Un checkpoint garantit que toutes les données sont vidées au moment où il se termine, mais le connecteur n'attend pas que l'intervalle entier soit écoulé avant de vider le tampon. Le connecteur déclenche une vidange anticipée si certaines conditions de tampon sont remplies (pour plus d'informations, consultez les rubriques Hologres, Hologres et Hologres). Étant donné qu'un entrepôt de données ne nécessite généralement pas de cohérence transactionnelle, le connecteur vide les données de manière asynchrone en arrière-plan. Il effectue ensuite une vidange forcée finale lors de chaque checkpoint pour préparer la reprise après incident.

Le déploiement du job génère l'erreur permission denied for database****

• Cause

  À partir de Realtime Compute for Apache Flink VVR 8.0.4, le connecteur impose le mode JDBC pour consommer le journal binaire des instances Hologres V2.0 ou ultérieures. Pour ces instances, les comptes non superutilisateur nécessitent des permissions spéciales pour consommer le journal binaire en mode JDBC.

• Solution

  Accordez au compte non superutilisateur les permissions nécessaires pour consommer le journal binaire en mode JDBC.

  user_name fait référence à un ID de compte Alibaba Cloud ou à un utilisateur RAM. Pour plus d'informations, consultez la rubrique Présentation des comptes.

  -- For the expert permission model, grant the CREATE permission and the replication role to the user.
  GRANT CREATE ON DATABASE <db_name> TO <user_name>;
  alter role <user_name> replication;
  -- If the database uses the simple permission model (SPM), you cannot run GRANT statements. 
  -- Instead, use spm_grant to grant the user the Admin role for the database. You can also grant permissions directly in HoloWeb.
  call spm_grant('<db_name>_admin', '<user_name>');
  alter role <user_name> replication;

Échec de la reprise du job : l'ID de table analysé à partir du checkpoint diffère de l'ID de table actuel

• Cause

  Cette exception survient dans les versions VVR 8.0.5 à VVR 8.0.8 de Realtime Compute for Apache Flink. Lorsqu'un job utilisant une table source binlog Hologres reprend à partir d'un checkpoint, le moteur effectue une vérification stricte de l'ID de table. Si l'ID actuel de la table Hologres ne correspond pas à celui stocké dans le checkpoint, la reprise échoue. Cela indique que la table source a été tronquée ou recréée pendant l'exécution du job.

• Solution

  Effectuez une mise à niveau vers la version VVR 8.0.9 ou ultérieure, puis redémarrez le job. La version VVR 8.0.9 supprime la vérification stricte de l'ID de table afin de prendre en charge des scénarios métier complexes. Évitez toutefois de recréer une table source binlog. En effet, la recréation d'une table efface l'intégralité de son historique binlog. Si Flink utilise ensuite le décalage de consommation de l'ancienne table pour lire les données de la nouvelle, cela risque d'entraîner des incohérences de données.

Précision inattendue des données Binlog en mode JDBC

• Cause

  Dans les versions 8.0.10 et antérieures de Realtime Compute for Apache Flink, une précision de données inattendue se produit si la précision du type DECIMAL déclarée dans une DDL Flink pour une table source Binlog ne correspond pas à celle définie dans Hologres.

• Solution

  Ce problème est corrigé dans la version 8.0.11 de Realtime Compute for Apache Flink. Assurez-vous néanmoins que la précision du type DECIMAL reste cohérente entre Flink et Hologres afin d'éviter toute perte de précision.

La suppression et la recréation d'une table portant le même nom peuvent provoquer les exceptions no table is defined in publication ou The table xxx has no slot named xxx dans les jobs

• Cause

  Ce phénomène s'explique par le fait que la suppression de la table n'entraîne pas la suppression de sa publication associée.

• Solution

  Solution 1 : Dans Hologres, exécutez l'instruction select * from pg_publication where pubname not in (select pubname from pg_publication_tables); pour identifier les publications résiduelles des tables supprimées. Exécutez ensuite l'instruction drop publication xx; pour les supprimer. Redémarrez enfin le job.

  Solution 2 : Utilisez la version VVR 8.0.5 ou ultérieure. Le connecteur gère automatiquement le nettoyage.

ClassCastException lors de la lecture depuis Hologres

• Détails de l'erreur

  Le message d'erreur ressemble à ce qui suit :

  java.lang.ClassCastException: class java.lang.Integer cannot be cast to class java.lang.Long (java.lang.Integer and java.lang.Long are in module java.base of loader 'bootstrap')

• Cause

  Cette erreur apparaît lorsque le type d'un champ dans la DDL Flink ne correspond pas au type du champ associé dans la table physique Hologres. Par exemple, un champ peut être défini comme BIGINT dans la DDL Flink alors que le champ correspondant dans la table Hologres est de type INTEGER. Étant donné que la vérification des types est ignorée pour les valeurs NULL, le job ne déclenche l'exception qu'au moment de la lecture effective des données.

• Solution

  Consultez la documentation Hologres Résumé des types de données pour vous assurer que les types de champs de votre DDL Flink correspondent bien à ceux de la table physique Hologres.

LogSizeTooLargeException

• Détails de l'erreur

  Caused by: com.aliyun.openservices.aliyun.log.producer.errors.LogSizeTooLargeException: the logs is 8785684 bytes which is larger than MAX_BATCH_SIZE_IN_BYTES 8388608
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.ensureValidLogSize(LogAccumulator.java:249)
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.doAppend(LogAccumulator.java:103)
  at com.aliyun.openservices.aliyun.log.producer.internals.LogAccumulator.append(LogAccumulator.java:84)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:385)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:308)
  at com.aliyun.openservices.aliyun.log.producer.LogProducer.send(LogProducer.java:211)
  at com.alibaba.ververica.connectors.sls.sink.SLSOutputFormat.writeRecord(SLSOutputFo
  rmat.java:100)

• Cause

  Cette erreur se produit lorsqu'un journal sur une seule ligne envoyé à Log Service dépasse la limite de taille de 8 Mo.

• Solution

  Pour ignorer l'entrée de journal trop volumineuse, modifiez la position de départ. Pour plus de détails, consultez la rubrique Démarrage du job.

OOM TaskManager : erreur d'espace de tas Java lors de la reprise

• Cause

  Ce problème résulte généralement d'un corps de message SLS trop volumineux. Le connecteur SLS demande les données par lots. Le nombre de LogGroups par lot est contrôlé par le paramètre batchGetSize, dont la valeur par défaut est 100. Ainsi, chaque requête peut récupérer jusqu'à 100 LogGroups. En fonctionnement normal, le programme Flink consomme les données rapidement et récupère rarement un lot complet de 100 LogGroups. Cependant, lors d'un basculement, une grande quantité de données non consommées peut s'accumuler. Si la mémoire nécessaire pour traiter un lot complet de 100 LogGroups excède la mémoire disponible de la JVM, le TaskManager subit une erreur OOM.

• Solution

  Réduisez la valeur du paramètre batchGetSize.

Définir le décalage de consommation pour une table source Paimon

Pour définir le décalage de consommation d'une table source Paimon, utilisez le paramètre scan.mode. Le tableau suivant décrit les valeurs disponibles et leur comportement.

Configurer l'expiration automatique des partitions

Les tables Paimon peuvent supprimer automatiquement les partitions dont la durée de vie dépasse le délai d'expiration spécifié. Cette fonctionnalité permet de réduire les coûts de stockage. Le processus est le suivant :

• Durée de vie : différence entre l'heure système actuelle et l'horodatage dérivé de la valeur de partition. La conversion de la valeur de partition en horodatage s'effectue comme suit :

  1. Convertissez une valeur de partition en chaîne temporelle à l'aide de la chaîne de format spécifiée par le paramètre partition.timestamp-pattern.

     Dans cette chaîne de format, une colonne de partition est représentée par un signe dollar ($) suivi du nom de la colonne. Par exemple, si les colonnes de partition sont year, month, day et hour, la chaîne de format $year-$month-$day $hour:00:00 convertit la partition year=2023,month=04,day=21,hour=17 en la chaîne 2023-04-21 17:00:00.

  2. Convertissez la chaîne temporelle en horodatage à l'aide de la chaîne de format spécifiée par le paramètre partition.timestamp-formatter.

     Si ce paramètre n'est pas défini, le système utilise par défaut les formats yyyy-MM-dd HH:mm:ss et yyyy-MM-dd. Vous pouvez utiliser toute chaîne de format compatible avec le DateTimeFormatter de Java.

• Délai d'expiration de la partition : valeur configurée pour le paramètre partition.expiration-time.

Résoudre les problèmes de données manquantes dans le stockage

• Les données peuvent ne pas être immédiatement visibles dans le stockage. Un writer Flink vide les données sur le disque dans les conditions suivantes :

  • Les données mises en mémoire tampon dans un bucket atteignent une certaine taille (par défaut : 64 Mo).

  • La taille totale de la mémoire tampon atteint un seuil défini (par défaut : 1 Go).

  • Un checkpoint est déclenché, ce qui force la vidange de toutes les données présentes en mémoire.

• Si vous utilisez l'écriture en flux, assurez-vous que le mécanisme de checkpointing est activé.

Gérer les données dupliquées dans Hudi

• Pour les écritures COW, activez le paramètre write.insert.drop.duplicates.

  Par défaut, une écriture COW ne déduplique pas les données du premier fichier de chaque bucket et applique uniquement la déduplication aux données incrémentielles. Pour effectuer une déduplication globale, vous devez activer ce paramètre. Pour une écriture MOR, aucun paramètre supplémentaire n'est requis. La définition d'une clé primaire active la déduplication globale par défaut.

  Remarque

  À partir de la version 0.10.0 de Hudi, cette propriété est renommée write.precombine et prend la valeur true par défaut.

• Pour effectuer une déduplication sur plusieurs partitions, définissez le paramètre index.global.enabled sur true.

  Remarque

  • À partir de la version 0.10.0 de Hudi, cette propriété prend la valeur true par défaut.

  • Lorsque index.type=bucket, définir le paramètre index.global.enabled sur true est inefficace car les index Bucket ne prennent pas en charge les modifications inter-partitions. Par conséquent, même si l'index global est activé, la fonctionnalité de déduplication ne peut pas fonctionner sur plusieurs partitions.

• Pour les mises à jour sur fenêtre longue, telles que la modification de données vieilles d'un mois, augmentez la valeur du paramètre index.state.ttl, exprimée en jours.

  L'index constitue la structure de données centrale de Hudi pour identifier les doublons. Le paramètre index.state.ttl contrôle la durée de conservation de l'état de l'index. La valeur par défaut précédente était de 1,5 jour. Une valeur inférieure ou égale à 0 indique que l'état de l'index est conservé indéfiniment.

  Remarque

  À partir de la version 0.10.0 de Hudi, cette propriété prend la valeur 0 par défaut.

Merge On Read ne contient que des fichiers journaux

• Cause : Hudi crée des fichiers Parquet uniquement après une compaction ; dans le cas contraire, il ne génère que des fichiers journaux. Par défaut, les tables Merge On Read utilisent une compaction asynchrone qui déclenche un job de compaction tous les cinq commits.

• Solution : Ajustez le paramètre compaction.delta_commits pour déclencher les jobs de compaction plus tôt.

Erreur : « multi-statement be found »

• Problème

  Un job Flink qui écrit des données dans une instance AnalyticDB for MySQL (ADB) échoue et redémarre. Les journaux affichent une erreur semblable à la suivante : Caused by: java.sql.SQLSyntaxErrorException: [13000, 2024101216171419216823505703151806929] multi-statement be found.

  at java.util.TimerThread.run(Timer.java:505)
  Caused by: java.sql.BatchUpdateException: [13000, 2024101216400819216823505703151079281] multi-statement be found.
  	at sun.reflect.GeneratedConstructorAccessor115.newInstance(Unknown Source)
  	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  	at com.mysql.cj.util.Util.handleNewInstance(Util.java:192)
  	at com.mysql.cj.util.Util.getInstance(Util.java:167)
  	at com.mysql.cj.util.Util.getInstance(Util.java:174)
  	at com.mysql.cj.jdbc.exceptions.SQLError.createBatchUpdateException(SQLError.java:224)
  	at com.mysql.cj.jdbc.ClientPreparedStatement.executePreparedBatchAsMultiStatement(ClientPreparedStatement.java:584)
  	at com.mysql.cj.jdbc.ClientPreparedStatement.executeBatchInternal(ClientPreparedStatement.java:431)
  	at com.mysql.cj.jdbc.StatementImpl.executeBatch(StatementImpl.java:795)
  	at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.ProxyStatement.executeBatch(ProxyStatement.java:127)
  	at com.ververica.cdc.connectors.shaded.com.zaxxer.hikari.pool.HikariProxyPreparedStatement.executeBatch(HikariProxyPreparedStatement...)
  	at com.ververica.connectors.mysql.table.sink.MySqlOutputFormat.executeSql(MySqlOutputFormat.java:567)
  	... 6 more

• Cause

  Cette erreur signale un problème de compatibilité entre la version 8.x du pilote JDBC MySQL et une base de données AnalyticDB for MySQL (ADB) où l'option ALLOW_MULTI_QUERIES=true est activée.

• Solution

  1. Contactez le support technique pour obtenir un connecteur ADB 3.0 personnalisé utilisant la version 5.1.46 du pilote JDBC MySQL. Appliquez ce connecteur à votre tâche Flink. Consultez la rubrique Gérer les connecteurs personnalisés pour obtenir des instructions sur l'utilisation d'un connecteur personnalisé.

  2. Définissez le paramètre allowMultiQueries=true dans l'URI de la table ADB, par exemple : jdbc:mysql://xxxxx.ads.aliyuncs.com:3306/xxx?allowMultiQueries=true.

Erreur : No suitable driver found

• Cause

  Le connecteur personnalisé ne parvient pas à trouver le pilote requis.

• Solution

  • Chargez le pilote dans la classe factory en appelant Class.forName.

  • Ajoutez le pilote en tant que dépendance supplémentaire et définissez le paramètre kubernetes.application-mode.classpath.include-user-jar sur true. Consultez la rubrique Comment configurer les paramètres d'exécution de job personnalisés ? pour obtenir des instructions.

Perte ou écrasement de données lors de l'écriture Flink vers Elasticsearch

• **Cause 1 : Conflit entre doc_as_upsert et un Ingest Pipeline Elasticsearch**

  Lorsque les deux éléments suivants sont configurés simultanément, Elasticsearch peut traiter les mises à jour partielles et les transformations du pipeline dans un ordre incompatible, ce qui entraîne l'écrasement ou la perte inattendue de documents :

  • sink.bulk-flush.update.doc_as_upsert = 'true' dans la clause WITH de la DDL Flink

  • Un Ingest Pipeline Elasticsearch assigné à l'index cible

  Dans cette configuration, Elasticsearch applique l'Ingest Pipeline avant de traiter la mise à jour partielle. Selon la logique du pipeline et la version d'Elasticsearch, cette interaction peut provoquer l'écrasement des champs du document par des valeurs incorrectes ou la suppression silencieuse de documents.

• Cause 2 : Plusieurs tâches sink Flink écrivant dans le même index

  Si deux jobs Flink ou plus — ou plusieurs instances sink parallèles avec des plages de clés disjointes — écrivent dans le même index Elasticsearch sans coordination des clés, les documents risquent de s'écraser mutuellement. Cela provoque une perte de données pour le sink qui écrit en dernier sur un ID de document donné.

• Solutions

  • **Pour le conflit entre doc_as_upsert et l'Ingest Pipeline :** Supprimez le paramètre sink.bulk-flush.update.doc_as_upsert = 'true' de la DDL Flink, puis supprimez ou réassignez l'Ingest Pipeline Elasticsearch de l'index cible. Déplacez toute logique de transformation des données précédemment gérée par le pipeline vers le job Flink lui-même — par exemple, en utilisant une ProcessFunction ou des colonnes calculées avant l'opérateur sink.

  • Pour l'écrasement mutuel par plusieurs tâches sink : Assurez-vous que chaque job Flink écrit dans un index Elasticsearch distinct, ou regroupez les écritures dans un seul job Flink qui gère les clés de document de manière déterministe.