How to configure a new Checkpoint using test_yaml_config
This how-to guide demonstrates advanced examples for configuring a CheckpointThe primary means for validating data in a production deployment of Great Expectations. using test_yaml_config. Note: For a basic guide on creating a new Checkpoint, please see How to create a new Checkpoint.
test_yaml_config is a convenience method for configuring the moving parts of a Great Expectations deployment. It allows you to quickly test out configs for DatasourcesProvides a standard API for accessing and interacting with data from a wide variety of source systems., StoresA connector to store and retrieve information about metadata in Great Expectations., and Checkpoints. test_yaml_config is primarily intended for use within a notebook, where you can iterate through an edit-run-check loop in seconds.
Steps
1. Create a new Checkpoint
From the CLICommand Line Interface, execute:
great_expectations checkpoint new my_checkpoint
This will open a Jupyter Notebook with a framework for creating and saving a new Checkpoint. Run the cells in the Notebook until you reach the one labeled "Test Your Checkpoint Configuration."
2. Edit your Checkpoint
The Checkpoint configuration that was created when your Jupyter Notebook loaded uses an arbitrary BatchA selection of records from a Data Asset. of data and Expectation SuiteA collection of verifiable assertions about data. to generate a basic Checkpoint configuration in the second code cell. You can edit this configuration to point to add additional entries under the validations key, or to edit the existing one. You can even replace this configuration entirely.
In the Additional Information section at the end of this guide you will find examples of other Checkpoint configurations you can use as your starting point, as well as explanations of the various ways you can arrange the keys and values in your Checkpoint's config_yaml.
After you make edits to the yaml_config variable, don't forget to re-run the cell that contains it!
3. Use test_yamal_config() to validate your Checkpoint configuration
Once you have made changes to the yaml_config in your Jupyter Notebook, you can verify that the updated configuration is valid by running the following code:
my_checkpoint = context.test_yaml_config(yaml_config=yaml_config)
This code is found in the code cell under the "Test Your Checkpoint Configuration" in your Jupyter Notebook.
If your Checkpoint configuration is valid, you will see an output stating that your checkpoint was instantiated successfully, followed by a Python dictionary representation of the configuration yaml you edited.
4. (Optional) Repeat from step 2
From here you can continue to edit your Checkpoint. After each change you should re-run the cell that contains the edited config_yaml and then verify that your configuration remains valid by re-running test_yaml_config(...).
5. Save your edited Checkpoint
Once you have made all of the changes you planned to implement and your last test_yaml_config() check passed, you are ready to save the Checkpoint you've created. At this point, run the remaining cells in your Jupyter Notebook.
Your checkpoint will be saved by the cell that contains the command:
context.add_checkpoint(**yaml.load(yaml_config))
Additional Information
Example SimpleCheckpoint configuration
The SimpleCheckpoint class takes care of some defaults which you will need to set manually in the Checkpoints class. The following example shows all possible configuration options for SimpleCheckpoint:
config = """
name: my_simple_checkpoint
config_version: 1.0
class_name: SimpleCheckpoint
validations:
- batch_request:
datasource_name: data__dir
data_connector_name: my_data_connector
data_asset_name: TestAsset
data_connector_query:
index: 0
expectation_suite_name: yellow_tripdata_sample_2019-01.warning
site_names: my_local_site
slack_webhook: my_slack_webhook_url
notify_on: all # possible values: "all", "failure", "success"
notify_with: # optional list of DataDocs site names to display in Slack message
"""
Example Checkpoint configurations
If you require more fine-grained configuration options, you can use the Checkpoint base class instead of SimpleCheckpoint.
In this example, the Checkpoint configuration uses the nesting of batch_request sections inside the validations block so as to use the defaults defined at the top level.
config = """
name: my_fancy_checkpoint
config_version: 1
class_name: Checkpoint
run_name_template: "%Y-%M-foo-bar-template-$VAR"
validations:
- batch_request:
datasource_name: my_datasource
data_connector_name: my_special_data_connector
data_asset_name: users
data_connector_query:
index: -1
- batch_request:
datasource_name: my_datasource
data_connector_name: my_other_data_connector
data_asset_name: users
data_connector_query:
index: -2
expectation_suite_name: users.delivery
action_list:
- name: store_validation_result
action:
class_name: StoreValidationResultAction
- name: store_evaluation_params
action:
class_name: StoreEvaluationParametersAction
- name: update_data_docs
action:
class_name: UpdateDataDocsAction
evaluation_parameters:
param1: "$MY_PARAM"
param2: 1 + "$OLD_PARAM"
runtime_configuration:
result_format:
result_format: BASIC
partial_unexpected_count: 20
"""
The following Checkpoint configuration runs the top-level action_list against the top-level batch_request as well as the locally-specified action_list against the top-level batch_request.
config = """
name: airflow_users_node_3
config_version: 1
class_name: Checkpoint
batch_request:
datasource_name: my_datasource
data_connector_name: my_special_data_connector
data_asset_name: users
data_connector_query:
index: -1
validations:
- expectation_suite_name: users.warning # runs the top-level action list against the top-level batch_request
- expectation_suite_name: users.error # runs the locally-specified action_list union with the top-level action-list against the top-level batch_request
action_list:
- name: quarantine_failed_data
action:
class_name: CreateQuarantineData
- name: advance_passed_data
action:
class_name: CreatePassedData
action_list:
- name: store_validation_result
action:
class_name: StoreValidationResultAction
- name: store_evaluation_params
action:
class_name: StoreEvaluationParametersAction
- name: update_data_docs
action:
class_name: UpdateDataDocsAction
evaluation_parameters:
environment: $GE_ENVIRONMENT
tolerance: 0.01
runtime_configuration:
result_format:
result_format: BASIC
partial_unexpected_count: 20
"""
The Checkpoint mechanism also offers the convenience of templates. The first Checkpoint configuration is that of a valid Checkpoint in the sense that it can be run as long as all the parameters not present in the configuration are specified in the run_checkpoint API call.
config = """
name: my_base_checkpoint
config_version: 1
class_name: Checkpoint
run_name_template: "%Y-%M-foo-bar-template-$VAR"
action_list:
- name: store_validation_result
action:
class_name: StoreValidationResultAction
- name: store_evaluation_params
action:
class_name: StoreEvaluationParametersAction
- name: update_data_docs
action:
class_name: UpdateDataDocsAction
evaluation_parameters:
param1: "$MY_PARAM"
param2: 1 + "$OLD_PARAM"
runtime_configuration:
result_format:
result_format: BASIC
partial_unexpected_count: 20
"""
The above Checkpoint can be run using the code below, providing missing parameters from the configured Checkpoint at runtime.
checkpoint_run_result: CheckpointResult
checkpoint_run_result = data_context.run_checkpoint(
checkpoint_name="my_base_checkpoint",
validations=[
{
"batch_request": {
"datasource_name": "my_datasource",
"data_connector_name": "my_special_data_connector",
"data_asset_name": "users",
"data_connector_query": {
"index": -1,
},
},
"expectation_suite_name": "users.delivery",
},
{
"batch_request": {
"datasource_name": "my_datasource",
"data_connector_name": "my_other_data_connector",
"data_asset_name": "users",
"data_connector_query": {
"index": -2,
},
},
"expectation_suite_name": "users.delivery",
},
],
)
However, the run_checkpoint method can be simplified by configuring a separate Checkpoint that uses the above Checkpoint as a template and includes the settings previously specified in the run_checkpoint method:
config = """
name: my_fancy_checkpoint
config_version: 1
class_name: Checkpoint
template_name: my_base_checkpoint
validations:
- batch_request:
datasource_name: my_datasource
data_connector_name: my_special_data_connector
data_asset_name: users
data_connector_query:
index: -1
- batch_request:
datasource_name: my_datasource
data_connector_name: my_other_data_connector
data_asset_name: users
data_connector_query:
index: -2
expectation_suite_name: users.delivery
"""
Now the run_checkpoint method is as simple as in the previous examples:
checkpoint_run_result = context.run_checkpoint(
checkpoint_name="my_fancy_checkpoint",
)
The checkpoint_run_result in both cases (the parameterized run_checkpoint method and the configuration that incorporates another configuration as a template) are the same.
The final example presents a Checkpoint configuration that is suitable for the use in a pipeline managed by Airflow.
config = """
name: airflow_checkpoint
config_version: 1
class_name: Checkpoint
validations:
- batch_request:
datasource_name: my_datasource
data_connector_name: my_runtime_data_connector
data_asset_name: IN_MEMORY_DATA_ASSET
expectation_suite_name: users.delivery
action_list:
- name: store_validation_result
action:
class_name: StoreValidationResultAction
- name: store_evaluation_params
action:
class_name: StoreEvaluationParametersAction
- name: update_data_docs
action:
class_name: UpdateDataDocsAction
"""
To run this Checkpoint, the batch_request with the batch_data nested under the runtime_parameters attribute needs to be specified explicitly as part of the run_checkpoint() API call, because the data to be ValidatedThe act of applying an Expectation Suite to a Batch. is accessible only dynamically during the execution of the pipeline.
checkpoint_run_result: CheckpointResult = data_context.run_checkpoint(
checkpoint_name="airflow_checkpoint",
batch_request={
"runtime_parameters": {
"batch_data": my_data_frame,
},
"data_connector_query": {
"batch_filter_parameters": {
"airflow_run_id": airflow_run_id,
}
},
},
run_name=airflow_run_id,
)