Блог

Моя конфигурация источника данных выглядит так:

 datasource_config = {
    "name": "example_datasource",
    "class_name": "Datasource",
    "module_name": "great_expectations.datasource",
    "execution_engine": {
        "module_name": "great_expectations.execution_engine",
        "class_name": "PandasExecutionEngine",
    },
    "data_connectors": {
        "default_runtime_data_connector_name": {
            "class_name": "RuntimeDataConnector",
            "module_name": "great_expectations.datasource.data_connector",
            "batch_identifiers": ["default_identifier_name"],
        },
    },
}
context.add_datasource(**datasource_config)
 

Мой фрейм данных Pandas и batch_requests были успешно созданы следующими командами:

 ...
df = read_csv_pandas(file_path="../done/my_file.txt", 
                           sep="|", 
                           header=0,
                           quoting=csv.QUOTE_ALL)

batch_request = RuntimeBatchRequest(
datasource_name="example_datasource",
data_connector_name="default_runtime_data_connector_name",
data_asset_name="MyDataAsset",
runtime_parameters={"batch_data": df},
batch_identifiers={"default_identifier_name": "default_identifier"}
)
 

Мой номер ожидания:

 expectation_suite_name = "My_validations"
suite = context.create_expectation_suite(expectation_suite_name, overwrite_existing=True)

 

Затем я создаю валидатор.

 validator = context.get_validator(
    batch_request=batch_request, expectation_suite_name=expectation_suite_name
)
validator.head(2)

 

Последняя команда успешно выводит 2 строки моего фрейма данных.

Затем я добавляю ожидания в свой номер.

 validator.expect_table_columns_to_match_ordered_list(['last_name', 'first_name', 'sex'])
validator.expect_column_values_to_be_in_set("sex", ["male", "female", "other", "unknown"])
validator.save_expectation_suite(discard_failed_expectations=False)
 

Затем я создаю документы с данными:

 suite_identifier = ExpectationSuiteIdentifier(expectation_suite_name=expectation_suite_name)
context.build_data_docs(resource_identifiers=[suite_identifier])
context.open_data_docs(resource_identifier=suite_identifier)
 

My checkpoint looks like:

 name: my_checkpoint_2
config_version: 1
class_name: SimpleCheckpoint
validations:
    - batch_request:
        datasource_name: example_datasource
        data_connector_name: default_runtime_data_connector_name
        data_asset_name: MyDataAsset
        runtime_parameters:
          batch_data: {df}
        batch_identifiers:
          default_identifier_name: default_identifier
expectation_suite_name: My_validations
 

Но эта команда

 context.run_checkpoint(checkpoint_name="my_checkpoint_2")
 

выдает ошибку:

 ValueError: RuntimeDataBatchSpec must provide a Pandas DataFrame or PandasBatchData object.