Les E/S gérées permettent à Dataflow de gérer des connecteurs d'E/S spécifiques utilisés dans les pipelines Apache Beam. Les E/S gérées simplifient la gestion des pipelines qui s'intègrent aux sources et aux récepteurs compatibles.
L'E/S gérée se compose de deux composants qui fonctionnent ensemble:
Transformation Apache Beam qui fournit une API commune pour créer des connecteurs d'E/S (sources et récepteurs).
Un service Dataflow qui gère ces connecteurs d'E/S en votre nom, y compris la possibilité de les mettre à niveau indépendamment de la version d'Apache Beam.
Les avantages de l'E/S gérée sont les suivants:
Mises à niveau automatiques Dataflow met automatiquement à niveau les connecteurs d'E/S gérés de votre pipeline. Cela signifie que votre pipeline reçoit des correctifs de sécurité, des améliorations des performances et des corrections de bugs pour ces connecteurs, sans avoir à modifier le code. Pour en savoir plus, consultez la section Mises à niveau automatiques.
API cohérente Traditionnellement, les connecteurs d'E/S d'Apache Beam disposent d'API distinctes, et chaque connecteur est configuré différemment. Les E/S gérées fournissent une seule API de configuration qui utilise des propriétés clé-valeur, ce qui permet d'obtenir un code de pipeline plus simple et plus cohérent. Pour en savoir plus, consultez la section API de configuration.
Conditions requises
Les SDK suivants sont compatibles avec les E/S gérées:
- SDK Apache Beam pour Java version 2.58.0 ou ultérieure.
- Le SDK Apache Beam pour Python version 2.61.0 ou ultérieure.
Le service backend nécessite Dataflow Runner v2. Si Runner V2 n'est pas activé, votre pipeline s'exécute toujours, mais il ne bénéficie pas des avantages du service d'E/S géré.
Mises à niveau automatiques
Les pipelines Dataflow avec des connecteurs d'E/S gérés utilisent automatiquement la dernière version fiable du connecteur, comme suit:
Lorsque vous envoyez une tâche, Dataflow utilise la dernière version du connecteur qui a été testée et fonctionne correctement.
Pour les tâches de streaming, Dataflow recherche des mises à jour chaque fois que vous lancez une tâche de remplacement et utilise automatiquement la dernière version fonctionnelle connue. Dataflow effectue cette vérification même si vous ne modifiez aucun code dans la tâche de remplacement.
Vous n'avez pas besoin de mettre à jour manuellement le connecteur ni la version Apache Beam de votre pipeline.
Le schéma suivant illustre le processus de mise à niveau. L'utilisateur crée un pipeline Apache Beam à l'aide de la version X du SDK. Lorsque l'utilisateur envoie la tâche, Dataflow vérifie la version de l'E/S gérée et la met à niveau vers la version Y.
Le processus de mise à niveau ajoute environ deux minutes au temps de démarrage d'une tâche. Pour vérifier l'état des opérations d'E/S gérées, recherchez les entrées de journal qui incluent la chaîne "Managed Transform(s)".
API de configuration
L'E/S gérée est une transformation Apache Beam clé en main qui fournit une API cohérente pour configurer des sources et des récepteurs.
Pour créer une source ou un puits compatible avec les E/S gérées, vous devez utiliser la classe Managed en Java. Spécifiez la source ou le récepteur à instancier, puis transmettez un ensemble de paramètres de configuration, comme suit:
Map config = ImmutableMap.<String, Object>builder()
.put("config1", "abc")
.put("config2", 1);
pipeline.apply(Managed.read(/*Which source to read*/).withConfig(config))
.getSinglePCollection();
Vous pouvez également transmettre des paramètres de configuration sous la forme d'un fichier YAML. Pour obtenir un exemple de code complet, consultez Lire à partir d'Apache Iceberg.
Destinations dynamiques
Pour certains collecteurs, le connecteur d'E/S géré peut sélectionner dynamiquement une destination en fonction des valeurs de champ dans les enregistrements entrants.
Pour utiliser des destinations dynamiques, fournissez une chaîne de modèle pour la destination. La chaîne de modèle peut inclure des noms de champ entre accolades, par exemple "tables.{field1}". Au moment de l'exécution, le connecteur remplace la valeur du champ pour chaque enregistrement entrant afin de déterminer la destination de cet enregistrement.
Par exemple, supposons que vos données comportent un champ nommé airport. Vous pouvez définir la destination sur "flights.{airport}". Si airport=SFO, l'enregistrement est écrit dans flights.SFO. Pour les champs imbriqués, utilisez la notation par points. Par exemple : {top.middle.nested}.
Filtrage
Vous pouvez filtrer certains champs avant qu'ils ne soient écrits dans la table de destination. Pour les embases compatibles avec les destinations dynamiques, vous pouvez utiliser le paramètre drop, keep ou only à cette fin. Ces paramètres vous permettent d'inclure des métadonnées de destination dans les enregistrements d'entrée, sans les écrire à la destination.
Vous ne pouvez définir qu'un seul de ces paramètres pour un collecteur donné.
| Paramètre de configuration | Type de données | Description |
|---|---|---|
drop |
liste de chaînes | Liste des noms de champs à supprimer avant d'écrire à la destination. |
keep |
liste de chaînes | Liste des noms de champs à conserver lors de l'écriture dans la destination. Les autres champs sont supprimés. |
only |
chaîne | Nom d'un seul champ à utiliser comme enregistrement de niveau supérieur à écrire lors de l'écriture dans la destination. Tous les autres champs sont supprimés. Ce champ doit être de type ligne. |
Sources et récepteurs compatibles
Les E/S gérées sont compatibles avec les sources et les récepteurs suivants.