Manage Checkpoints
A Checkpoint validates Expectation Suite data. After you create a Checkpoint to validate data, you can save and reuse the Checkpoint.
To learn more about Checkpoints, see Checkpoint.
Prerequisites
-
You have deployed the GX Agent. See Deploy the GX Agent.
-
You have a Data Asset.
-
You have created an Expectation.
Add a Checkpoint
-
Run the following code to import the
great_expectations
module and the existing Data Context:Pythonimport great_expectations as gx
context = gx.get_context() -
Run the following code to retrieve the Expectation Suite:
Pythonexpectation_suite = context.get_expectation_suite(expectation_suite_name=<expectation_name>)
-
Run the following code to assign a name to the Checkpoint:
Pythoncheckpoint_name = <checkpoint_name>
-
Run the following code to define the Checkpoint configuration.
Pythoncheckpoint_config = {
"name": checkpoint_name,
"validations": [{
"expectation_suite_name": expectation_suite.expectation_suite_name,
"expectation_suite_ge_cloud_id": expectation_suite.ge_cloud_id,
"batch_request": {
"datasource_name": "<data_source_name>",
"data_asset_name": "<data_asset_name>",
},
}],
}Replace
data_source_name
anddata_asset_name
with the names of an existing Data Source and Data Asset. If you haven't connected to a Data Source and created a Data Asset, see Manage Data Assets. -
Run the following code to add the Checkpoint:
Pythoncheckpoint = context.add_or_update_checkpoint(**checkpoint_config)
-
Optional. Run the following code to confirm the Checkpoint name:
Pythonprint(checkpoint)
-
Optional. Run the following code to run the Checkpoint:
Pythonresult = checkpoint.run()
Run a Checkpoint
-
In GX Cloud, click Checkpoints.
-
Optional. To run a Checkpoint on a failing Checkpoint, click Failures Only.
-
Optional. To run a specific Checkpoint, select it in the Checkpoints pane.
-
Click Run Checkpoint for the Checkpoint you want to run.
Add a Validation to a Checkpoint
Add validation data to a Checkpoint to aggregate individual Expectation Suite or Data Source Validations into a single Checkpoint. For more information, see Add Validation data or Expectation Suites to a Checkpoint in the GX OSS documentaion.
-
In GX Cloud, click Checkpoints.
-
Click Edit Checkpoint in the Checkpoints list for the Checkpoint you want to add the Validation.
-
Copy the code snippet and then close the Edit Checkpoint dialog.
-
Paste the code snippet into a Python interpreter and then add the following code block:
Pythonvalidations = [
{
"batch_request": {
"datasource_name": "your_datasource_name",
"data_asset_name": "your_data_asset_name",
},
"expectation_suite_name": "your.expectation.suite.name",
},
]Replace
your_datasource_name
,your_data_asset_name
, andyour.expectation.suite.name
with your own values. -
Optional. Repeat step 4 to add additional Validations.
-
Run the following code to update the Checkpoint configuration:
Pythoncheckpoint = context.add_or_update_checkpoint(**checkpoint_config)
Edit a Checkpoint name
-
In GX Cloud, click Checkpoints.
-
Click Edit Checkpoint in the Checkpoints list for the Checkpoint you want to edit.
-
Enter a new name for the Checkpoint and then click Save.
-
Update the Checkpoint name in all code that included the previous Checkpoint name.
Edit a Checkpoint configuration
-
Run the following code to import the
great_expectations
module and the existing Data Context:Pythonimport great_expectations as gx
context = gx.get_context() -
In GX Cloud, click Checkpoints.
-
Click Edit Checkpoint in the Checkpoints list for the Checkpoint you want to edit.
-
Copy the code snippet and then close the Edit Checkpoint dialog.
-
Paste the the code snippet into a Python interpreter and then edit the Checkpoint configuration.
-
Run the following code to update the Checkpoint configuration:
Pythoncheckpoint = context.add_or_update_checkpoint(**checkpoint_config)
Configure the Checkpoint result format parameter
You can use the result_format
parameter to define the level of detail you want returned with your Validation Results. For example, you can return a success or failure message, a summary of observed values, a list of failing values, or you can add a query or a filter function that returns all failing rows. For more information, see Result format.
Run the following code to apply result_format
to every Expectation in a Suite:
checkpoint: Checkpoint = Checkpoint(
name="my_checkpoint",
run_name_template="%Y%m%d-%H%M%S-my-run-name-template",
data_context=context,
batch_request=my_batch_request,
expectation_suite_name="test_suite",
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"}},
],
runtime_configuration={
"result_format": {
"result_format": "COMPLETE",
"unexpected_index_column_names": ["pk_column"],
"return_unexpected_index_query": True,
},
},
)
Replace my_checkpoint
and test_suite
with your own values. You define your Checkpoint configuration below the runtime_configuration
key. The results are stored in the Validation Result after you run the Checkpoint.
Delete a Checkpoint
-
In GX Cloud, click Checkpoints.
-
Click Delete Checkpoint for the Checkpoint you want to delete.
-
Click Delete.