Remplissage des données historiques avec des pipelines

Dans l’ingénierie des données, le remplissage fait référence au 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 de diffusion en continu.

En règle générale, il s’agit d’un flux distinct qui envoie des données dans vos tables existantes. L’illustration suivante présente un flux de renvoi de données historiques vers les tables bronze de votre pipeline.

Remplissage rétroactif pour ajouter des données historiques à un flux de travail existant

Certains scénarios qui peuvent nécessiter un remblai :

  • Traitez les données historiques d’un système hérité pour entraîner un modèle Machine Learning (ML) ou créer un tableau de bord d’analyse de tendance historique.
  • Retraitez un sous-ensemble de données en raison d’un problème de qualité des données liées aux sources de données en amont.
  • Vos besoins métier ont changé et vous devez renvoyer des données pour une période différente qui n’a pas été couverte par le pipeline initial.
  • Votre logique métier a changé et vous devez retraiter les données historiques et actuelles.

Un remplissage rétroactif dans les pipelines Lakeflow est pris en charge par un flux d’ajout spécialisé qui utilise l’option ONCE. Pour plus d’informations sur l’option, consultez append_flow ou ONCE

Considérations relatives au remplissage des données historiques dans une table de flux

  • En règle générale, ajoutez les données à la table de streaming bronze. Les couches d’argent et d’or en aval récupèrent les nouvelles données de la couche bronze.
  • Vérifiez que votre pipeline peut gérer correctement les données dupliquées au cas où les mêmes données sont ajoutées plusieurs fois.
  • Vérifiez que le schéma de données historiques est compatible avec le schéma de données actuel.
  • Tenez compte de la taille du volume de données et du contrat SLA de temps de traitement requis, et configurez en conséquence les tailles de cluster et de lot.

Exemple : ajout d’un remplissage à un pipeline existant

Dans cet exemple, supposons que vous disposez d’un pipeline qui ingère des données brutes d’inscription d’événements à partir d’une source de stockage cloud, à compter du 01 janvier 2025. Vous réalisez ultérieurement que vous souhaitez compléter rétroactivement les trois années précédentes de données historiques pour les besoins d'analyse et de rapports en aval. Toutes les données se trouvent à un emplacement, partitionnés par année, mois et jour, au format JSON.

Pipeline initiale

Voici le code de pipeline de démarrage qui ingère de façon incrémentielle les données d’inscription d’événements brutes à partir du stockage cloud.

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

-- 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 Chargeur automatique pour nous assurer que nous ne traitons pas toutes les données à partir du chemin de stockage cloud. Le traitement incrémentiel est arrêté à cette frontière.

Conseil / Astuce

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

Remplissage des données des 3 dernières années

Vous souhaitez maintenant ajouter un ou plusieurs flux pour recharger les données précédentes. Dans cet exemple, procédez comme suit :

  • Utilisez le append once flux. Cela effectue un renvoi unique sans continuer à s’exécuter après ce premier renvoi. Le code reste dans votre pipeline et, si le pipeline est un jour entièrement actualisé, le renvoi est réexécuté.
  • Créez trois flux de renvoi, 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 dans 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 calcul sans serveur, vous pouvez ajuster le nombre maximum de travailleurs pour le pipeline. En augmentant le nombre maximal de workers, vous vous assurez de disposer des ressources nécessaires pour traiter les données historiques tout en continuant à traiter les données en streaming actuelles en respectant le SLA.

Conseil / Astuce

Si vous utilisez le calcul serverless avec une mise à l’échelle automatique améliorée (valeur par défaut), votre cluster augmente automatiquement quand votre charge augmente.

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

-- 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 responsabilités

  • Le traitement incrémentiel est indépendant des opérations de remplissage.
  • Chaque flux a ses propres paramètres de configuration et d’optimisation.
  • Il existe une distinction claire entre les opérations incrémentielles et de remplissage.

Exécution contrôlée

  • L’utilisation de l’option ONCE garantit que chaque remplissage s’exécute exactement une fois.
  • Le flux de renvoi reste dans le graphique de pipeline, mais devient inactif une fois terminé. Il est prêt à être utilisé lors de l’actualisation complète, automatiquement.
  • Il existe une piste d’audit claire des opérations de renvoi dans la définition du pipeline.

Optimisation du traitement

  • Vous pouvez diviser le large lot de données à remplir en plusieurs lots plus petits pour un traitement plus rapide ou pour mieux contrôler le processus.
  • L'utilisation de l'autoscaling amélioré adapte dynamiquement la taille du cluster en fonction de sa charge actuelle.

Évolution du schéma

  • L'utilisation de schemaEvolutionMode="addNewColumns" gère les changements de schéma sans problème.
  • Vous disposez d’une inférence de schéma cohérente entre les données historiques et actuelles.
  • Il existe 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 une table de dimension à changement lent (SCD) qui existe déjà dans un système hérité avec des années d’histoire accumulée, mais dont le flux de modifications original n’est plus disponible. Comme les événements de modification d’origine ont disparu, vous relisez à la place une seule fois l’historique propre à l’ancienne table vers la nouvelle cible AUTO CDC, puis attachez un nouveau flux CDC pour la suite. Pour en savoir plus sur AUTO CDC et les types de SCD, voir Les API AUTO CDC : Simplifier la capture des données de modifications avec pipelines.

Le schéma consiste en un flux ponctuel AUTO CDC vers la même table de streaming que celle ciblée par le flux continu AUTO CDC. Une cible AUTO CDC n’accepte que les flux AUTO CDC ; la graine doit donc aussi être un flux AUTO CDC. Un flux d’ajout simple INSERT INTO ONCE dans la même table ne passe pas la validation :

  1. Créez la table de streaming cible dans laquelle votre flux AUTO CDC écrit.
  2. Amorcez l’historique existant une seule fois à l’aide d’un flux AUTO CDC ONCE qui lit la table SCD existante en tant que flux, séquencé selon la colonne de début de validité de l’ancienne table. Relisez les lignes héritées en tant qu’événements de modification au lieu 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 ces colonnes directement.
  3. Attachez le flux AUTO CDC continu en lisant le nouveau flux de modification. AUTO CDC résout le classement par clé, donc le basculement doit être respecté individuellement pour chaque clé d’entreprise : la première modification en direct de chaque clé doit être séquencée après la dernière modification amorcée pour cette même clé. Une valeur de séquence légèrement supérieure au maximum global hérité peut rester obsolète pour une clé individuelle, et le premier changement en temps réel de cette clé est alors ignoré ou mal ordonné.

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

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 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 selon un horodatage, qui utilise un seul instant de bascule pour séparer l’historique amorcé du flux en direct. Si la table héritée est séquencée selon une valeur d’un type différent de celui du flux en direct, effectuer un cast de l’un des deux afin que les types correspondent.

La même structure fonctionne pour une cible SCD de type 1 : remplacer STORED AS SCD TYPE 2 par STORED AS SCD TYPE 1 dans les deux flux, et la cible ne conserve que la ligne actuelle par clé. Avant de vous appuyer sur l’une ou l’autre configuration, validez, sur un échantillon de clés, que la première modification en production d’une clé initialisée produit exactement une nouvelle version et clôture correctement la version précédente. Un écart de séquençage par clé apparaît généralement à cette étape.

Ressources supplémentaires

  • Last updated on