AWS Data pipeline - comment l'utiliser pour les mises à jour de données RDS incrémentielles?

Question

J'ai une situation où j'utilise le pipeline de données pour importer des données du fichier csv stocké dans S3. Pour la charge de données initiale, le pipeline de données s'exécute correctement.

Maintenant, je dois garder cette base de données à jour et synchronisée avec la base de données interne. Ce qui signifie qu'il y aura un ensemble de fichiers CSV venant à S3 qui seraient les mises à jour de certains enregistrements existants, de nouveaux enregistrements ou de la suppression. J'ai besoin de cela pour être mis à jour sur RDS grâce à un pipeline de données. Question - Le pipeline de données peut-il être conçu à cette fin OU est-il uniquement destiné à une charge de données ponctuelle? S'il peut être utilisé pour des mises à jour incrémentielles, comment puis-je y remédier.

Toute aide est très appréciée!

Answer 1

Oui, vous devez faire une mise à jour et insérer (alias upsert).

Si vous avez une table avec les touches: key_a, key_b et autres colonnes: col_c, col_d vous pouvez utiliser l'instruction SQL suivante:

insert into TABLENAME (key_a, key_b, col_c, col_d) values (?,?,?,?) ON DUPLICATE KEY UPDATE col_c=values(col_c), col_d=values(col_d)

Answer 2

Veuillez se référer à la documentation SSFE: http://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-template-incrementalcopyrdstos3.html

Il existe un modèle prédéfini pour le téléchargement progressif Mysql RDS, j'ai personnellement essayé de téléchargements incrémentiels MySQL, SQL Server et redshift. Vous pouvez commencer par utiliser le modèle mysql et l'éditer en vue architecte pour avoir un aperçu des nouvelles/additions supplémentaires qu'il utilise et également créer des données pour d'autres bases de données RDS.

En interne, le incrémental vous oblige à fournir la colonne de changement qui doit être essentiellement une colonne de date, et ce changecolumn eux est utilisé dans le script Sql qui ressemble à:

select * from # {table} où # {myRDSTableLastModifiedCol}> = '# {format (@scheduledStartTime,' AAAA-MM-jj HH-mm-ss ')}' et # {myRDSTableLastModifiedCol} < = '# {format (@scheduledEndTime,' AAAA-MM-jj HH-mm-ss '}'

scheduledStartTime et scheduleEndTime sont l'expression de la ligne de données dont la valeur dépend de votre planification. Et scheduletype est une série de temps pour exécuter le sql à la fin de l'heure de fin de planification afin de garantir qu'il n'y a pas de perte de données.

Et oui, les données supprimées ne peuvent pas être suivies via la ligne de données; aussi datapipleline n'aiderait pas non plus si la colonne datetime n'est pas là dans votre table, auquel cas je préférerais charger la table complète.

J'espère avoir assez-bien couvert je sais :)

Cordialement, Varun R