Log

Progettare pipeline di forecasting per la produzione

Rami Krispin

Senior Manager, Data Science and Engineering

Log della pipeline

Logging

Metadati
Test di validazione
Stato
Performance

Log della pipeline

Logging

Metadati
Test di validazione
Stato
Performance

Funzionalità

Osservabilità
Debug
Decisioni per ruolo

Impostare i log

Imposta lo schema del log
Gestisci scenari:
- Nessun nuovo dato disponibile
- Esito della validazione
Accoda il log

Il processo di log ETL

Verifica attività API

import etl.eia_etl as ee

log = ee.Log()

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

log.create_log(facets = facets)

Schema log

Il processo di log ETL

Verifica attività API

import etl.eia_etl as ee

log = ee.Log()

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

log.create_log(facets = facets)

Schema log

Il processo di log ETL

Verifica attività API

import etl.eia_etl as ee

log = ee.Log()

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

log.create_log(facets = facets)

Schema log

Il processo di log ETL

Verifica attività API

import etl.eia_etl as ee

log = ee.Log()

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

log.create_log(facets = facets)

Schema log

Il processo di log ETL

Verifica attività API

import etl.eia_etl as ee

log = ee.Log()

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

log.create_log(facets = facets)

Schema log

Il processo di log ETL

Verifica attività API

import etl.eia_etl as ee

log = ee.Log()

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

log.create_log(facets = facets)

Schema log

Il processo di log ETL

Verifica attività API

import etl.eia_etl as ee

log = ee.Log()

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

log.create_log(facets = facets)

Schema log

Il processo di log ETL

Stato API

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

Stato API

Il processo di log ETL

Stato dati

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

Stato API

Il processo di log ETL

Attività di validazione

if not meta.updates_available:
  log.no_updates()

Stato API

Il processo di log ETL

Attività di validazione

Stato API

Il processo di log ETL

Aggiorna dati

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)

Output dati

Il processo di log ETL

Aggiornamento dati non riuscito

log.failure()

Il processo di log ETL

Validazione dati

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)

Il processo di log ETL

Validazione dati

Riepilogo validazione

Il processo di log ETL

Validazione non riuscita

log.failure()

Il processo di log ETL

Accoda i dati

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