
message = checker(results)

validation_pipeline()As can be seen from the step definition , the step takes in a pandas.DataFrame dataset and a boolean condition and it returns a Great Expectations CheckpointResult object. The boolean condition is only used as a means of ordering steps in a pipeline (e.g. if you must force it to run only after the data profiling step generates an Expectation Suite):

@step

def great_expectations_validator_step(

dataset: pd.DataFrame,

expectation_suite_name: str,

data_asset_name: Optional[str] = None,

action_list: Optional[List[Dict[str, Any]]] = None,

exit_on_error: bool = False,

) -> CheckpointResult:

You can view the complete list of configuration parameters in the SDK docs.

Call Great Expectations directly

You can use the Great Expectations library directly in your custom pipeline steps, while leveraging ZenML's capability of serializing, versioning and storing the ExpectationSuite and CheckpointResult objects in its Artifact Store. To use the Great Expectations configuration managed by ZenML while interacting with the Great Expectations library directly, you need to use the Data Context managed by ZenML instead of the default one provided by Great Expectations, e.g.:

import great_expectations as ge

from zenml.integrations.great_expectations.data_validators import (

GreatExpectationsDataValidator

import pandas as pd

from great_expectations.core import ExpectationSuite

from zenml import step

@step

def create_custom_expectation_suite(

) -> ExpectationSuite:

"""Custom step that creates an Expectation Suite

Returns:

An Expectation Suite

"""

context = GreatExpectationsDataValidator.get_data_context()

# instead of:

# context = ge.get_context()

expectation_suite_name = "custom_suite"

suite = context.create_expectation_suite(

expectation_suite_name=expectation_suite_name

expectation_configuration = ExpectationConfiguration(...)

suite.add_expectation(expectation_configuration=expectation_configuration)

...

context.save_expectation_suite(