Ingestione dati

Progettare pipeline di forecasting per la produzione

Rami Krispin

Senior Manager, Data Science and Engineering

Processo ETL

Processo ETL - sorgente

Processo di ingestione dati

Processo ETL - estrazione

Processo di ingestione dati

Processo ETL - trasformazione

Processo di ingestione dati

Processo ETL - caricamento

Processo di ingestione dati

Processo ETL - validazione dati

Processo di ingestione dati

Processo ETL - integrità dei dati

Processo di ingestione dati

Processo ETL - logging

Processo di ingestione dati

Processo ETL - refresh

Processo di ingestione dati

Processo ETL - refresh

Processo di ingestione dati

Processo ETL - refresh

Processo di ingestione dati

Processo ETL - refresh

Processo di ingestione dati

Processo ETL - avvio nuova pipeline

Processo di ingestione dati

Processo ETL - aggiornare i dati normalizzati

Processo di ingestione dati

Processo ETL - verifica parità dei dati

Task Check API

Processo ETL - metadati

Task Data Status

Processo ETL - refresh dei dati

Task Data Refresh

Processo ETL - validazione dati

Task Data Validation

Processo ETL - analisi dati

Task Analisi validazione dati

Processo ETL - aggiungi i dati

Task Append the Data

Processo ETL - refresh previsione

Task Refresh the Forecast

Processo ETL - valutazione previsione

Task Score the Forecast

Processo ETL - nessun aggiornamento

Task Validazione

Processo ETL - errore di refresh

Task Validazione

Processo ETL - errore di validazione

Task Validazione

Controlli di validazione dati

Schema dati
Range valori
Valori mancanti
Duplicati
Logica di business

Controlli di validazione dati

print(raw)

period    respondent    respondent-name    type    type-name    value    value-units
0    2025-05-01 00:00:00    US48    United States Lower 48    D    Demand    504242    
1    2025-04-30 23:00:00    US48    United States Lower 48    D    Demand    508099    
2    2025-04-30 22:00:00    US48    United States Lower 48    D    Demand    508323    
3    2025-04-30 21:00:00    US48    United States Lower 48    D    Demand    500551    
4    2025-04-30 20:00:00    US48    United States Lower 48    D    Demand    492240

Controlli di validazione dati

import pointblank as pb

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

Controlli di validazione dati

validation = (pb.Validate(data=raw,
              tbl_name="US48 Data Validation",
              label="Data Refresh",
              thresholds=pb.Thresholds(warning=0.2, error=0, critical=0.1))

.col_schema_match(schema=table_schema)

.col_vals_gt(columns="value", value=0)

.col_vals_in_set(columns="respondent", set = ["US48"])
.col_vals_in_set(columns="type", set = ["D"])

.col_vals_not_null(columns=["period", "value"])

.rows_distinct()

.interrogate())

Controlli di validazione dati

Validazione dati

Controlli di validazione dati

print(validation.all_passed())

True

¡Vamos a practicar!

Progettare pipeline di forecasting per la produzione