Remplissage rétroactif des données historiques avec des pipelines

En Data Engineering, le remplissage rétrospectif désigne le processus de traitement rétroactif des données historiques par le biais d'un pipeline de données conçu pour traiter les données actuelles ou en streaming.

Généralement, il s’agit d’un flux distinct qui envoie des données dans vos tables existantes. L'illustration suivante montre un flux de remplissage historique envoyant des données historiques aux tables bronze de votre pipeline.

Voici quelques scénarios qui pourraient nécessiter un remplissage :

• Traiter les données historiques d'un système hérité pour entraîner un modèle de Machine Learning (ML) ou créer un tableau de bord d'analyse des tendances historiques.
• Retraiter un sous-ensemble de données en raison d'un problème de qualité des données avec des sources de données en amont.
• Vos besoins commerciaux ont changé et vous devez effectuer un remplissage rétroactif des données pour une période différente qui n'était pas couverte par le pipeline initial.
• La logique de votre entreprise a changé et vous devez retraiter les données historiques et actuelles.

Un rattrapage dans les LakeFlow Pipelines est pris en charge par un flux d’ajout spécialisé qui utilise l’option ONCE. Consultez append_flow ou CREATE FLOW (pipelines) pour plus d’informations sur l’option ONCE.

Considérations lors du remplissage des données historiques dans une table de streaming

• Généralement, ajoutez les données à la table de streaming bronze. Les couches Silver et Gold en aval reprendront les nouvelles données de la couche Bronze.
• Assurez-vous que votre pipeline peut gérer les données en double avec élégance si les mêmes données sont ajoutées plusieurs fois.
• Assurez-vous que le schéma des données historiques est compatible avec le schéma des données actuelles.
• Considérez la taille du volume de données et le SLA de temps de traitement requis, et configurez en conséquence les tailles de clusters et de batch.

Exemple : Ajouter un remplissage à un pipeline existant

Dans cet exemple, supposons que vous ayez un pipeline qui ingère les données brutes d'enregistrement d'événements à partir d'une source de stockage cloud, à partir du 01.01.2025. Vous réalisez ensuite que vous souhaitez remplir les trois années précédentes de données historiques pour les cas d'usage d'analyse et de rapports en aval. Toutes les données se trouvent dans un seul emplacement, partitionnées par année, mois et jour, au format JSON.

Pipeline initial

Voici le code de pipeline de départ qui ingère de manière incrémentale les données brutes d'enregistrement d'événements à partir du stockage cloud.

Python

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
incremental_load_path = f"{source_root_path}/*/*/*"

# create a streaming table and the default flow to ingest streaming events
@dp.table(name="registration_events_raw", comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2025-01-01T00:00:00.000+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}") # safeguard to not process data before begin_year
    )

SQL

SQL

-- create a streaming table and the default flow to ingest streaming events
CREATE OR REFRESH STREAMING LIVE TABLE registration_events_raw AS
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025'; -- safeguard to not process data before begin_year

Ici, nous utilisons l'option modifiedAfter Auto Loader pour nous assurer que nous ne traitons pas toutes les données à partir du chemin de stockage cloud. Le traitement incrémentiel est interrompu à cette limite.

astuce

D'autres sources de données, telles que Kafka, Kinesis et Azure Event Hub, ont des options de lecture équivalentes pour obtenir le même comportement.

Remplir les données des 3 années précédentes.

Vous souhaitez maintenant ajouter un ou plusieurs flux pour un remplissage des données précédentes. Dans cet exemple, suivez les étapes suivantes :

• Utilisez le flux append once. Cela effectue un rattrapage unique sans continuer à s'exécuter après ce premier rattrapage. Le code reste dans votre pipeline, et si le pipeline est entièrement actualisé, le rattrapage est réexécuté.
• Créez trois flux de remplissage, un pour chaque année (dans ce cas, les données sont divisées par année dans le chemin). Pour Python, nous paramétrons la création des flux, mais en SQL nous répétons le code trois fois, une fois pour chaque flux.

Si vous travaillez sur votre propre projet et que vous n'utilisez pas de compute serverless, vous pouvez mettre à jour le nombre maximum de Workers pour le pipeline. L'augmentation du nombre maximal de Workers garantit que vous disposez des Ressources pour traiter les données historiques tout en continuant de traiter les données de streaming actuelles dans le respect du SLA attendu.

astuce

Si vous utilisez le Serverless compute avec l'autoscaling amélioré (le default), votre cluster augmente automatiquement en taille lorsque votre charge augmente.

Python

Python

from pyspark import pipelines as dp

source_root_path = spark.conf.get("registration_events_source_root_path")
begin_year = spark.conf.get("begin_year")
backfill_years = spark.conf.get("backfill_years") # e.g. "2024,2023,2022"
incremental_load_path = f"{source_root_path}/*/*/*"

# meta programming to create append once flow for a given year (called later)
def setup_backfill_flow(year):
    backfill_path = f"{source_root_path}/year={year}/*/*"
    @dp.append_flow(
        target="registration_events_raw",
        once=True,
        name=f"flow_registration_events_raw_backfill_{year}",
        comment=f"Backfill {year} Raw registration events")
    def backfill():
        return (
            spark
            .read
            .format("json")
            .option("inferSchema", "true")
            .load(backfill_path)
        )

# create the streaming table
dp.create_streaming_table(name="registration_events_raw", comment="Raw registration events")

# append the original incremental, streaming flow
@dp.append_flow(
        target="registration_events_raw",
        name="flow_registration_events_raw_incremental",
        comment="Raw registration events")
def ingest():
    return (
        spark
        .readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.maxFilesPerTrigger", 100)
        .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
        .option("modifiedAfter", "2024-12-31T23:59:59.999+00:00")
        .load(incremental_load_path)
        .where(f"year(timestamp) >= {begin_year}")
    )

# parallelize one time multi years backfill for faster processing
# split backfill_years into array
for year in backfill_years.split(","):
    setup_backfill_flow(year) # call the previously defined append_flow for each year

SQL

SQL

-- create the streaming table
CREATE OR REFRESH STREAMING TABLE registration_events_raw;

-- append the original incremental, streaming flow
CREATE FLOW
  registration_events_raw_incremental
AS INSERT INTO
  registration_events_raw BY NAME
SELECT * FROM STREAM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/*/*/*",
  format => "json",
  inferColumnTypes => true,
  maxFilesPerTrigger => 100,
  schemaEvolutionMode => "addNewColumns",
  modifiedAfter => "2024-12-31T23:59:59.999+00:00"
)
WHERE year(timestamp) >= '2025';


-- one time backfill 2024
CREATE FLOW
  registration_events_raw_backfill_2024
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2024/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2023
CREATE FLOW
  registration_events_raw_backfill_2023
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2023/*/*",
  format => "json",
  inferColumnTypes => true
);

-- one time backfill 2022
CREATE FLOW
  registration_events_raw_backfill_2022
AS INSERT INTO ONCE
  registration_events_raw BY NAME
SELECT * FROM read_files(
  "/Volumes/gc/demo/apps_raw/event_registration/year=2022/*/*",
  format => "json",
  inferColumnTypes => true
);

Cette implémentation met en évidence plusieurs modèles importants.

Séparation des préoccupations

• Le traitement incrémentiel est indépendant des Opérations de rattrapage.
• Chaque flux dispose de sa propre configuration et de ses propres paramètres d'optimisation.
• Il existe une distinction claire entre les Opérations incrémentielles et les Opérations de remplissage.

Exécution contrôlée

• L'utilisation de l'option ONCE garantit que chaque remplissage rétroactif s'exécute exactement une seule fois.
• Le flux de remplissage rétrospectif reste dans le graphe de pipeline, mais devient inactif une fois terminé. Il est prêt à être utilisé lors du full refresh, automatiquement.
• Il existe une piste d'audit claire des opérations de remplissage des données dans la définition du pipeline.

Optimisation du traitement

• Vous pouvez diviser le grand backfill en plusieurs backfills plus petits pour un traitement plus rapide, ou pour contrôler le traitement.
• L'autoscaling amélioré redimensionne dynamiquement la taille du cluster en fonction de la charge actuelle du cluster.

évolution des schémas

• L'utilisation de schemaEvolutionMode="addNewColumns" gère les modifications de schéma avec élégance.
• Vous disposez d'une inférence de schéma cohérente pour les données historiques et actuelles.
• Il y a une gestion sécurisée des nouvelles colonnes dans les données plus récentes.

Exemple : Remblai une cible SCD lors d’une migration

Un scénario de migration courant est celui d'une table de dimensions à évolution lente (SCD) qui existe déjà dans un système hérité avec des années d'historique accumulé, mais dont le flux de modification d'origine n'est plus disponible. Comme les événements de modification d'origine ont disparu, vous rejouez plutôt l'historique de la table héritée dans la nouvelle cible AUTO CDC une seule fois, puis vous attachez un nouveau flux CDC pour la suite. Pour en savoir plus sur AUTO CDC et les types de SCD, consultez The AUTO CDC APIs: Simplify change data capture with pipelines.

Le modèle est un flux AUTO CDC unique vers la même table de streaming que celle ciblée par le flux AUTO CDC en cours. Une cible AUTO CDC n'accepte que les flux AUTO CDC, le seed doit donc également être un flux AUTO CDC. Un flux d'ajout INSERT INTO ONCE simple vers la même table échoue à la validation :

1. Créez la table de streaming cible vers laquelle votre flux AUTO CDC écrit.
2. Initialisez l'historique hérité une fois avec un flux AUTO CDC ONCE qui lit la table SCD héritée en tant que Stream, séquencé par la colonne de start de validité héritée. Relisez les lignes héritées en tant qu'événements de changement plutôt que de les mettre en forme vous-même. AUTO CDC génère les colonnes d'historique __START_AT et __END_AT pour une cible SCD de type 2 ; n'écrivez donc pas directement dans ces colonnes.
3. Attachez le flux AUTO CDC en cours qui lit le flux de données de modification récent. AUTO CDC résout le classement par clé, de sorte que la transition doit être valide pour chaque clé métier individuellement : le premier changement en direct de chaque clé doit être séquencé après le dernier changement initialisé pour cette même clé. Une valeur de séquence qui est simplement postérieure au maximum global hérité peut toujours être obsolète pour une clé individuelle, et le premier changement en direct de cette clé est alors ignoré ou classé de manière incorrecte.

Le code suivant crée une table de streaming qui suit les étapes ci-dessus :

SQL

CREATE OR REFRESH STREAMING TABLE customers_history;

-- One-time seed: replay the legacy history as change events
CREATE FLOW customers_history_seed
AS AUTO CDC ONCE INTO customers_history
FROM stream(legacy.customers_scd2)
KEYS (customer_id)
SEQUENCE BY valid_from
STORED AS SCD TYPE 2;

-- Ongoing live CDC into the same target
CREATE FLOW customers_history_cdc
AS AUTO CDC INTO customers_history
FROM stream(customers_cdc_bronze)
KEYS (customer_id)
SEQUENCE BY change_timestamp
STORED AS SCD TYPE 2;

Les deux flux doivent s'accorder sur leurs clés, leur type de SCD et le type de données de leur colonne de séquençage. Dans l'exemple précédent, les deux flux sont séquencés par un timestamp, qui utilise une heure de basculement unique pour séparer l'historique amorcé du flux en direct. Si la table héritée est séquencée par une valeur d'un type différent de celui du flux en direct, effectuez un cast de l'une d'entre elles afin que les types correspondent.

La même forme fonctionne pour une cible SCD de type 1 : remplacez STORED AS SCD TYPE 2 par STORED AS SCD TYPE 1 sur les deux flux, et la cible ne conservera que la ligne actuelle par clé. Avant de vous appuyer sur l'une ou l'autre forme, validez sur un échantillon de clés que le premier changement en direct pour une clé amorcée produit exactement une nouvelle version et ferme correctement la précédente. Un écart de séquençage par clé apparaît généralement à cette étape.

Ressources supplémentaires

• Charger et traiter les données de manière incrémentielle avec LakeFlow Pipelines
• Utilisez les flux dans les LakeFlow Pipelines.
• append_flow
• CRÉER UN FLUX (pipelines)