Data ingestion

Designing Forecasting Pipelines for Production

Rami Krispin

Senior Manager, Data Science and Engineering

ETL process

ETL Process

ETL process - source

Data Ingestion Process

ETL process - extract

Data Ingestion Process

ETL process - transform

Data Ingestion Process

ETL process - load

Data Ingestion Process

ETL process - data validation

Data Ingestion Process

ETL process - data integrity

Data Ingestion Process

ETL process - logging

Data Ingestion Process

ETL process - refreshing

Data Ingestion Process

ETL process - refreshing

Data Ingestion Process

ETL process - refreshing

Data Ingestion Process

ETL process - refreshing

Data Ingestion Process

ETL process - triggering a new pipeline

Data Ingestion Process

ETL process - updating normalized data

Data Ingestion Process

ETL process - data parity check

Check API Task

ETL process - metadata

Data Status Task

ETL process - data refresh

Data Refresh Task

ETL process - data validation

Data Validation Task

ETL process - data analysis

Analyze Data Validation Task

ETL process - append the data

Append the Data Task

ETL process - forecast refresh

Refresh the Forecast Task

ETL process - forecast evaluation

Score the Forecast Task

ETL process - no update

Validation Task

ETL process - refresh failure

Validation Task

ETL process - validation failure

Validation Task

Data validation checks

Data schema
Values range
Missing values
Duplications
Business logic

Data validation checks

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

Data validation checks

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")
    ]
    )

Data validation checks

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())

Data validation checks

Data Validation

Data validation checks

print(validation.all_passed())

True

Let's practice!

Designing Forecasting Pipelines for Production