Journaux

Concevoir des pipelines de prévision pour la production

Rami Krispin

Senior Manager, Data Science and Engineering

Journaux de pipeline

Journalisation

Métadonnées
Tests de validation
Statut
Performances

Journaux de pipeline

Journalisation

Métadonnées
Tests de validation
Statut
Performances

Fonctionnalités

Observabilité
Débogage
Décisions selon les rôles

Configuration des journaux

Définir le schéma du journal
Gérer les cas :
- Aucune nouvelle donnée
- Résultat de la validation
Ajouter au journal

Le processus de journal ETL

Tâche de vérification de l'API

import etl.eia_etl as ee

log = ee.Log()

facet = {'respondent': 'US48', 
         'type': 'D'}

log.create_log(facets = facets)

Schéma du journal

Le processus de journal ETL

Tâche de vérification de l'API

import etl.eia_etl as ee

log = ee.Log()

facet = {'respondent': 'US48', 
         'type': 'D'}

log.create_log(facets = facets)

Schéma du journal

Le processus de journal ETL

Tâche de vérification de l'API

import etl.eia_etl as ee

log = ee.Log()

facet = {'respondent': 'US48', 
         'type': 'D'}

log.create_log(facets = facets)

Schéma du journal

Le processus de journal ETL

Tâche de vérification de l'API

import etl.eia_etl as ee

log = ee.Log()

facet = {'respondent': 'US48', 
         'type': 'D'}

log.create_log(facets = facets)

Schéma du journal

Le processus de journal ETL

Tâche de vérification de l'API

import etl.eia_etl as ee

log = ee.Log()

facet = {'respondent': 'US48', 
         'type': 'D'}

log.create_log(facets = facets)

Schéma du journal

Le processus de journal ETL

Tâche de vérification de l'API

import etl.eia_etl as ee

log = ee.Log()

facet = {'respondent': 'US48', 
         'type': 'D'}

log.create_log(facets = facets)

Schéma du journal

Le processus de journal ETL

Tâche de vérification de l'API

import etl.eia_etl as ee

log = ee.Log()

facet = {'respondent': 'US48', 
         'type': 'D'}

log.create_log(facets = facets)

Schéma du journal

Le processus de journal ETL

Tâche d'état des données

meta = ee.get_metadata(api_key = api_key,
api_path = api_meta_path, 
meta_path= log_path, 
facets = facets, 
offset = 22, window = 336)

Statut de l'API

Le processus de journal ETL

Tâche d'état des données

meta = ee.get_metadata(api_key = api_key,
api_path = api_meta_path, 
meta_path= log_path, 
facets = facets, 
offset = 22, window = 336)

Statut de l'API

Le processus de journal ETL

Tâche de validation

if not meta.updates_available:
  log.no_updates()

Statut de l'API

Le processus de journal ETL

Tâche de validation

Statut de l'API

Le processus de journal ETL

Tâche d'actualisation des données

get = ea.eia_get(api_key=api_key, 
    api_path= api_data_path, 
    data = "value", facets = facets, 
    start = meta.start, 
    end = meta.api_end_offset)

Sortie des données

Le processus de journal ETL

Échec de l'actualisation des données

log.failure()

Le processus de journal ETL

Tâche de validation des données

test = ee.Validation(data = get.data,  
tbl_name= "get request", 
label = "validation",
parameters=get.parameters,initial= False,
warning=0.10, error=0, critical=0)

schema_refresh = pb.Schema(
    columns=[
        ("index", "datetime64[ns]"),   
        ("respondent", "object"),
        ("respondent-name", "object"),
        ("type", "object"),
        ("type-name", "object"),
        ("value", "int64"),
        ("value-units", "object")
    ])

test.add_schema(schema = schema_refresh)

Le processus de journal ETL

Tâche de validation des données

Récapitulatif de validation des données

Le processus de journal ETL

Échec de la validation

log.failure()

Le processus de journal ETL

Ajouter les données

if log.log["status"]:
  print("Appending the data")
  df = ee.AppendData()
  df.append_data(data_path = data_path, 
    new_data = get.data,save = True,
    schema = schema_append, 
    parameters = get.parameters)
  print(df.validation)
  if df.status and df.save:
    log.log["update"] = True
  else:
    log.log["update"] = False