Utilisation de Kafka pour l'intégration de données avec mises à jour et suppressions

Question

Pour rappel, nous disposons d'un grand nombre de sources de données allant des SGBDR aux fichiers S3. Nous aimerions synchroniser et intégrer ces données avec d'autres entrepôts de données, bases de données, etc.

Au début, cela semblait être le modèle canonique de Kafka. Nous aimerions diffuser les changements de données à travers Kafka vers les sources de sortie de données. Dans notre cas de test, nous capturons les changements avec Oracle Golden Gate et les poussons avec succès vers une file d'attente Kafka. Cependant, pousser ces changements jusqu'à la source de sortie des données s'est avéré difficile.

Je me rends compte que cela fonctionnerait très bien si nous ne faisions qu'ajouter de nouvelles données aux sujets et aux files d'attente Kafka. Nous pourrions mettre en cache les modifications et les écrire dans les différentes sources de données. Mais ce n'est pas le cas. Nous allons mettre à jour, supprimer, modifier des partitions, etc. La logique pour gérer cela semble être beaucoup plus compliquée.

Nous avons essayé d'utiliser des tables d'étapes et des jointures pour mettre à jour/supprimer les données, mais j'ai l'impression que cela deviendrait rapidement très lourd.

J'en viens à ma question : y a-t-il d'autres façons d'aborder ces opérations ? Ou devrions-nous nous engager dans une autre direction ?

Toute suggestion ou aide est très appréciée. Je vous remercie de votre attention.

Answer 1

Vous pouvez adopter trois approches :

1. Chargement complet de la benne
2. Chargement incrémentiel de la décharge
3. Réplication du Binlog

Chargement complet de la benne

Périodiquement, vous devez enregistrer votre table source de données SGBDR dans un fichier et le charger dans l'entrepôt de données, en remplaçant la version précédente. Cette approche est surtout utile pour les petites tables, mais elle est très simple à mettre en œuvre et prend facilement en charge les mises à jour et les suppressions de données.

Chargement incrémentiel de la décharge

Périodiquement, vous récupérez les enregistrements qui ont changé depuis votre dernière requête et vous les envoyez dans l'entrepôt de données. Quelque chose comme

SELECT *
FROM my_table
WHERE last_update > #{last_import}

Cette approche est légèrement plus complexe à mettre en œuvre, car il faut maintenir l'état ("last_import" dans l'extrait ci-dessus), et elle ne prend pas en charge les suppressions. Elle peut être étendue pour prendre en charge les suppressions, mais cela la rend plus compliquée. Un autre inconvénient de cette approche est qu'elle exige que vos tables aient une propriété last_update colonne.

Réplication du Binlog

Écrivez un programme qui écoute en permanence le binlog de votre SGBDR et envoie ces mises à jour pour qu'elles soient chargées dans une table intermédiaire de l'entrepôt de données, contenant les valeurs mises à jour de la ligne et indiquant s'il s'agit d'une opération de suppression ou d'une opération de mise à jour/création. Rédigez ensuite une requête qui consolide périodiquement ces mises à jour afin de créer une table qui reflète la table d'origine. L'idée sous-jacente à ce processus de consolidation est de sélectionner, pour chaque identifiant, la dernière version (la plus avancée) telle qu'elle apparaît dans toutes les mises à jour ou dans la version précédente de la table consolidée.

Cette approche est légèrement plus complexe à mettre en œuvre, mais elle permet d'obtenir des performances élevées même sur des tables de grande taille et prend en charge les mises à jour et les suppressions.

Kafka est pertinent pour cette approche dans la mesure où il peut être utilisé comme pipeline pour les mises à jour des lignes entre l'auditeur binlog et le chargement dans la table intermédiaire de l'entrepôt de données.

---

Vous pouvez en savoir plus à ce sujet différentes approches de réplication dans cet article de blog .

Divulgation : je travaille à Alooma (un collègue a écrit l'article de blog lié ci-dessus, et nous fournissons des pipelines de données en tant que service, en résolvant des problèmes comme celui-ci).