Guide de compatibilité d'état pour les mises à niveau de Flink 2.2 - Service géré pour Apache Flink
Comprendre les modifications de compatibilité des étatsRéférence de compatibilité de sérialisationMéthodes diagnostiquesListe de contrôle préalable à la mise à niveauMigration d'ÉtatBonnes pratiquesÉtapes suivantes

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Guide de compatibilité d'état pour les mises à niveau de Flink 2.2

Lors de la mise à niveau de Flink 1.x vers Flink 2.2, des problèmes de compatibilité d'état peuvent empêcher la restauration de votre application à partir de snapshots. Ce guide vous aide à identifier les problèmes de compatibilité potentiels et propose des stratégies de migration.

Comprendre les modifications de compatibilité des états

Amazon Managed Service pour Apache Flink 2.2 introduit plusieurs modifications de sérialisation qui affectent la compatibilité des états. Les principaux sont les suivants :

  • Mise à niveau de la version Kryo : Apache Flink 2.2 met à niveau le sérialiseur Kryo fourni de la version 2 à la version 5. Comme Kryo v5 utilise un format de codage binaire différent de celui de Kryo v2, aucun état d'opérateur sérialisé via Kryo dans un point de sauvegarde Flink 1.x ne peut pas être restauré dans Flink 2.2.

  • Sérialisation des collections Java : Dans Flink 1.x, les collections Java (telles que HashMapArrayList, etHashSet) POJOs étaient sérialisées à l'aide de Kryo. Flink 2.2 introduit des sérialiseurs optimisés spécifiques à la collection qui sont incompatibles avec l'état sérialisé KRYO de la version 1.x. Les applications utilisant des collections Java avec des sérialiseurs POJO ou Kryo dans la version 1.x ne peuvent pas restaurer cet état dans Flink 2.2. Consultez la documentation de Flink pour plus de détails sur les types de données et la sérialisation.

  • Compatibilité du connecteur Kinesis : la version du connecteur Kinesis Data Streams (KDS) inférieure à 5.0 conserve un état incompatible avec le connecteur Flink 2.2 Kinesis version 6.0. Vous devez migrer vers la version 5.0 ou supérieure du connecteur avant de procéder à la mise à niveau.

Référence de compatibilité de sérialisation

Passez en revue toutes les déclarations d'état de votre application et associez les types de sérialisation au tableau ci-dessous. Si un type d'état est incompatible, consultez la Migration d'État section avant de procéder à la mise à niveau.

Référence de compatibilité de sérialisation
Type de sérialisation Compatible ? Détails
Avril (SpecificRecord,GenericRecord) Oui Utilise son propre format binaire indépendant de Kryo. Assurez-vous d'utiliser les informations de type Avro natives de Flink, et non d'Avro enregistré en tant que sérialiseur Kryo.
Protobuf Oui Utilise son propre codage binaire indépendant de Kryo. Vérifiez que les modifications du schéma respectent les règles d'évolution rétrocompatibles.
POJOs sans collections Oui Géré par le sérialiseur POJO de Flink, mais uniquement si la classe répond à tous les critères POJO : classe publique, constructeur public no-arg, tous les champs publics ou accessibles via des getters/setters, et tous les types de champs eux-mêmes sérialisables par Flink. Un POJO qui enfreint l'une de ces règles revient silencieusement à Kryo et devient incompatible.
Personnalisé TypeSerializers Oui Compatible uniquement si votre sérialiseur ne délègue pas à Kryo en interne.
État des API SQL et Table Oui (avec réserve) Utilise les sérialiseurs internes de Flink. Cependant, Apache Flink ne garantit pas la compatibilité des états entre les versions principales des applications de l'API Table. Testez d'abord dans un environnement hors production.
POJOs avec des collections Java (HashMap,ArrayList,HashSet) Non Dans Flink 1.x, les collections qu'il contient POJOs étaient sérialisées via Kryo v2. Flink 2.2 introduit des sérialiseurs de collection dédiés dont le format binaire est incompatible avec le format Kryo v2.
Classes de cas Scala Non Sérialisé via Kryo dans Flink 1.x. La mise à niveau de Kryo v2 vers v5 modifie le format binaire.
Enregistrements Java Non Revenez généralement à la sérialisation Kryo dans Flink 1.x. Vérifiez en testant avecdisableGenericTypes().
Types de bibliothèques tierces Non Les types sans sérialiseur personnalisé enregistré sont renvoyés à Kryo. Le changement de format binaire Kryo v2 à v5 rompt la compatibilité.
Tout type utilisant Kryo fallback Non Si Flink ne peut pas gérer un type avec un sérialiseur intégré ou enregistré, il revient à Kryo. Tous les états sérialisés par Kryo à partir de la version 1.x sont incompatibles avec la version 2.2.

Méthodes diagnostiques

Vous pouvez identifier les problèmes de compatibilité des états de manière proactive en consultant les journaux des applications ou en les inspectant après l'opération de l'UpdateApplication API.

Identifiez la solution de secours Kryo dans votre application

Vous pouvez utiliser le modèle regex suivant dans vos journaux pour identifier la solution de repli Kryo dans votre application :

Class class (?<className>[^\s]+) cannot be used as a POJO type

Exemple de journal :

Class class org.apache.flink.streaming.connectors.kinesis.model.SequenceNumber
cannot be used as a POJO type because not all fields are valid POJO fields,
and must be processed as GenericType. Please read the Flink documentation on
"Data Types & Serialization" for details of the effect on performance and
schema evolution.

Si la mise à niveau échoue à l'aide de l' UpdateApplication API, les exceptions suivantes peuvent indiquer que vous rencontrez une incompatibilité d'état liée au sérialiseur :

IndexOutOfBoundsException

Caused by: java.lang.IndexOutOfBoundsException: Index 116 out of bounds for length 1
    at java.base/jdk.internal.util.Preconditions.outOfBounds(Unknown Source)
    at java.base/jdk.internal.util.Preconditions.outOfBoundsCheckIndex(Unknown Source)
    at java.base/jdk.internal.util.Preconditions.checkIndex(Unknown Source)
    at java.base/java.util.Objects.checkIndex(Unknown Source)
    at java.base/java.util.ArrayList.get(Unknown Source)
    at com.esotericsoftware.kryo.util.MapReferenceResolver.getReadObject(MapReferenceResolver.java:77)
    at com.esotericsoftware.kryo.Kryo.readReferenceOrNull(Kryo.java:923)
    ... 23 more

StateMigrationException (POJOSerializer)

Caused by: org.apache.flink.util.StateMigrationException: The new state serializer
(org.apache.flink.api.java.typeutils.runtime.PojoSerializer@8bf85b5d) must not be
incompatible with the old state serializer
(org.apache.flink.api.java.typeutils.runtime.PojoSerializer@3282ee3).

Liste de contrôle préalable à la mise à niveau

  • Passez en revue toutes les déclarations d'État figurant dans votre demande

  • Vérifiez POJOs auprès des collections (HashMap,ArrayList,HashSet)

  • Vérifiez les méthodes de sérialisation pour chaque type d'état

  • Créez une application de réplication de production et testez la compatibilité des états à l'aide de l' UpdateApplication API sur cette réplique

  • Si l'état est incompatible, sélectionnez une stratégie parmi Migration d'État

  • Activez la restauration automatique dans la configuration de votre application Flink de production

Migration d'État

Reconstruire l'état complet

Idéal pour les applications où l'état peut être reconstruit à partir des données sources.

Si votre application peut rétablir son état à partir des données sources :

  1. Arrêtez l'application Flink 1.x

  2. Passez à Flink 2.x avec un code mis à jour

  3. Commencez par SKIP_RESTORE_FROM_SNAPSHOT

  4. Autoriser l'application à rétablir son état

aws kinesisanalyticsv2 start-application \
    --application-name MyApplication \
    --run-configuration '{
        "ApplicationRestoreConfiguration": {
            "ApplicationRestoreType": "SKIP_RESTORE_FROM_SNAPSHOT"
        }
    }'

Bonnes pratiques

  1. Utilisez toujours Avro ou Protobuf pour les états complexes. Ils permettent l'évolution du schéma et sont indépendants de Kryo

  2. Évitez les collections dans POJOs — Utilisez le natif de Flink ListState et MapState à la place

  3. Testez la restauration de l'état localement : avant la mise à niveau de production, effectuez un test avec des instantanés réels

  4. Prenez fréquemment des instantanés, en particulier avant les mises à niveau majeures des versions

  5. Activer la restauration automatique : configurez votre application MSF pour qu'elle soit automatiquement annulée en cas de défaillance

  6. Documentez vos types d'états — Conservez la documentation de tous les types d'états et de leurs méthodes de sérialisation

  7. Surveillez la taille des points de contrôle : l'augmentation de la taille des points de contrôle peut indiquer des problèmes de sérialisation

Étapes suivantes

Planifiez votre mise à niveau : voirMise à niveau vers Flink 2.2 : guide complet.

Pour toute question ou problème lors de la migration, consultez le Support Résoudre les problèmes liés au service géré pour Apache Flink ou contactez le AWS Support.