Is your feature request related to a problem? Please describe.
Validating data on ingestion is both non-trivial and important. It's important since Machine Learning models that consuming those features are usually highly sensible to any shifts in the data. It's non-trivial since every data model is unique and creating general solution would be not very effective. Our current validation includes only type checks and not-null restrictions applied to entities columns. Even if we will extend those check - that still would be row level validation. What we need is to look on data from distance - in batches - to detect that some column has too many null values (eg, more than 5% on average) and other column has too many outliers (distribution shift). Those checks involves more complicated data transformation, that shouldn't be a part of Feast and instead should be implemented by domain experts.
Thus we need to give users ability to use custom code inside ingestion that will be responsible for such transformations & checks. It should receive data in controlled batches and be able to call specific libraries like great-expectations, tfdv, scipy.
Describe the solution you'd like
The solution consists of two parts:
- In Python SDK we create UDFs and pickle the code to pass it to IngestionJob.
1a. In POC we don't allow to pass arbitrary code but rather have some templates that wraps specific solutions (great-expectations, tfdv, etc)
- Inside Ingestion (only streaming in POC) we gather all columns in batches and pass it as single Pandas Dataframe to the UDF. UDF is being applied right after reading data from source and before saving to storage (before ingestion).
2a. At start we expect UDF to return single column that would mark valid/invalid rows with flag.
SDK side (great-expectations example):
you_training_df = client.get_historical_features(...)
from great_expectations.dataset import PandasDataset
ds = PandasDataset.from_dataset(you_training_df)
ds.expect_column_values_to_be_between('column', 0, 100)
expectations = ds.get_expectation_suite()
from feast.contrib.validation.ge import create_validation_udf, apply_validation
validation_udf = create_validation_udf('myUDF', expectations)
apply_validation(client, feature_table, validation_udf, validation_window_secs=30)
Inside ingestion Job
validationUDF: StructType -> BooleanType
val rowsAfterValidation = batchDF.withColumn("_isValid", validationUDF(struct(columns:_*)))
rowsAfterValidation
.filter("!_isValid")
.map(updateMetrics)
.writeToDeadLetter
...
.trigger(ProcessingTimeTrigger.create(config.ValidationWindow, TimeUnit.SECONDS))
.start
Additional context
Previous iteration RFC
Is your feature request related to a problem? Please describe.
Validating data on ingestion is both non-trivial and important. It's important since Machine Learning models that consuming those features are usually highly sensible to any shifts in the data. It's non-trivial since every data model is unique and creating general solution would be not very effective. Our current validation includes only type checks and not-null restrictions applied to entities columns. Even if we will extend those check - that still would be row level validation. What we need is to look on data from distance - in batches - to detect that some column has too many null values (eg, more than 5% on average) and other column has too many outliers (distribution shift). Those checks involves more complicated data transformation, that shouldn't be a part of Feast and instead should be implemented by domain experts.
Thus we need to give users ability to use custom code inside ingestion that will be responsible for such transformations & checks. It should receive data in controlled batches and be able to call specific libraries like
great-expectations,tfdv,scipy.Describe the solution you'd like
The solution consists of two parts:
1a. In POC we don't allow to pass arbitrary code but rather have some templates that wraps specific solutions (great-expectations, tfdv, etc)
2a. At start we expect UDF to return single column that would mark valid/invalid rows with flag.
SDK side (great-expectations example):
Inside ingestion Job
Additional context
Previous iteration RFC