DQ data storage on S3 only?

[jaina15]

Hi @jmcorreia ,

I was working on this and added a dq validator check. But when I am executing the code using the below acon I am getting the following error:

acon = {
    "input_specs": [
        {
            "spec_id": "brz_test",
            "read_type": "batch",
            "data_format": "delta",
            "db_table": "cat.test.brz_tab",
        }
    ],
    "transform_specs": [
        {
            "spec_id": "brz_transformation",
            "input_id": "brz_test",
            "transformers": [
                {
                    "function": "add_current_date",
                    "args": {
                        "output_col": "lh_created_on"
                    }
                },
                {
                    "function": "with_expressions",
                    "args": {
                        "cols_and_exprs": {
                            "bonus": "age*10",
                            "designation": "CASE WHEN age <= 22 THEN 'Junior ENG' "
                                        "WHEN age>22 and age <=27 THEN 'Senior ENG'"
                                        "ELSE 'Staff ENG' END"
                        }
                    }
                }
            ],
        }
    ],
    "dq_specs": [
        {
            "spec_id": "check_brz_transformation",
            "input_id": "brz_transformation",
            "dq_type": "validator",
            "result_sink_db_table": "cat.test.dq_checks",
            "fail_on_error": False,
            "dq_functions": [
                {
                    "function": "expect_column_values_to_not_be_null",
                    "args": {
                        "column": "name"
                    }
                },
                {
                    "function": "expect_column_values_to_be_between",
                    "args": {
                        "column": "age",
                        "min_value": 18,
                        "max_value": 32
                    },
                }
            ],
        },
    ],
    "output_specs": [
        {
            "spec_id": "slv_test",
            "input_id": "check_brz_transformation",
            "data_format": "delta",
            "write_type": "merge",
            "merge_opts": {
                "merge_predicate": """
                    new.name = current.name
                """,
            },
            "db_table": "cat.test.slv_tab"
        }
    ],
    "terminate_specs": [
        {
            "function": "optimize_dataset",
            "args": {
                "db_table": "cat.test.slv_tab"
            }
        }
    ]
}

Below is the error that I am getting:
Invalid store configuration: Please check the configuration of your TupleS3StoreBackend named expectations_S3_store. Exception was: 
 expected string or bytes-like object
Invalid store configuration: Please check the configuration of your TupleS3StoreBackend named validations_S3_store. Exception was: 
 expected string or bytes-like object
Invalid store configuration: Please check the configuration of your TupleS3StoreBackend named checkpoint_S3_store. Exception was: 
 expected string or bytes-like object
Invalid store configuration: Please check the configuration of your TupleS3StoreBackend named profiler_S3_store. Exception was: 
 expected string or bytes-like object


File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/lakehouse_engine/engine.py:30, in load_data(acon_path, acon)
     28 acon = ConfigUtils.get_acon(acon_path, acon)
     29 ExecEnv.get_or_create(app_name="data_loader", config=acon.get("exec_env", None))
---> 30 return DataLoader(acon).execute()

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/lakehouse_engine/algorithms/data_loader.py:229, in DataLoader.execute(self)
    227 except Exception as e:
    228     NotifierFactory.generate_failure_notification(self.terminate_specs, e)
--> 229     raise e
    231 return written_dfs

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/lakehouse_engine/algorithms/data_loader.py:221, in DataLoader.execute(self)
    219 transformed_dfs = self.transform(read_dfs)
    220 self._logger.info("Starting data quality stage...")
--> 221 validated_dfs = self.process_dq(transformed_dfs)
    222 self._logger.info("Starting write stage...")
    223 written_dfs = self.write(validated_dfs)

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/lakehouse_engine/algorithms/data_loader.py:162, in DataLoader.process_dq(self, data)
    160     if spec.cache_df:
    161         df_processed_df.cache()
--> 162     dq_processed_dfs[spec.spec_id] = DQFactory.run_dq_process(
    163         spec, df_processed_df
    164     )
    165 else:
    166     dq_processed_dfs[spec.spec_id] = df_processed_df

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/lakehouse_engine/dq_processors/dq_factory.py:85, in DQFactory.run_dq_process(cls, dq_spec, data)
     78 context.add_datasource(**cls._get_data_source_defaults(dq_spec))
     80 expectation_suite_name = (
     81     dq_spec.expectation_suite_name
     82     if dq_spec.expectation_suite_name
     83     else f"{dq_spec.spec_id}-{dq_spec.input_id}-{dq_spec.dq_type}"
     84 )
---> 85 context.add_or_update_expectation_suite(
     86     expectation_suite_name=expectation_suite_name
     87 )
     89 batch_request = cls._get_batch_request(dq_spec, data)
     91 # We only run the checkpoint for the Validator, as we don't want to allow
     92 # running this for all the expectations suggested by the Assistant.

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/great_expectations/core/usage_statistics/usage_statistics.py:260, in usage_statistics_enabled_method.<locals>.usage_statistics_wrapped_method(*args, **kwargs)
    257         args_payload = args_payload_fn(*args, **kwargs) or {}
    258         nested_update(event_payload, args_payload)
--> 260     result = func(*args, **kwargs)
    261     message["success"] = True
    262 except Exception:

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/great_expectations/data_context/data_context/abstract_data_context.py:3324, in AbstractDataContext.add_or_update_expectation_suite(self, expectation_suite_name, id, expectations, evaluation_parameters, data_asset_type, execution_engine_type, meta, expectation_suite)
   3312     expectation_suite = ExpectationSuite(
   3313         expectation_suite_name=expectation_suite_name,
   3314         data_context=self,
   (...)
   3320         meta=meta,
   3321     )
   3323 try:
-> 3324     existing = self.get_expectation_suite(
   3325         expectation_suite_name=expectation_suite.name
   3326     )
   3327 except gx_exceptions.DataContextError:
   3328     # not found
   3329     return self._add_expectation_suite(expectation_suite=expectation_suite)

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/great_expectations/data_context/data_context/abstract_data_context.py:3406, in AbstractDataContext.get_expectation_suite(self, expectation_suite_name, include_rendered_content, ge_cloud_id)
   3401 if include_rendered_content is None:
   3402     include_rendered_content = (
   3403         self._determine_if_expectation_suite_include_rendered_content()
   3404     )
-> 3406 if self.expectations_store.has_key(key):
   3407     expectations_schema_dict: dict = cast(
   3408         dict, self.expectations_store.get(key)
   3409     )
   3410     # create the ExpectationSuite from constructor

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/great_expectations/data_context/store/store.py:260, in Store.has_key(self, key)
    258 if self._use_fixed_length_key:
    259     return self._store_backend.has_key(key.to_fixed_length_tuple())
--> 260 return self._store_backend.has_key(key.to_tuple())

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/great_expectations/data_context/store/_store_backend.py:176, in StoreBackend.has_key(self, key)
    174 def has_key(self, key) -> bool:
    175     self._validate_key(key)
--> 176     return self._has_key(key)

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/great_expectations/data_context/store/tuple_store_backend.py:705, in TupleS3StoreBackend._has_key(self, key)
    704 def _has_key(self, key):
--> 705     all_keys = self.list_keys()
    706     return key in all_keys

File /local_disk0/.ephemeral_nfs/envs/pythonEnv-46e86f9d-e1d9-4025-b0b1-6ad382cc706c/lib/python3.9/site-packages/great_expectations/data_context/store/tuple_store_backend.py:621, in TupleS3StoreBackend.list_keys(self, prefix)
    618 def list_keys(self, prefix: Tuple = ()) -> List[Tuple]:
    619     # Note that the prefix arg is only included to maintain consistency with the parent class signature
    620     s3r = self._create_resource()
--> 621     bucket = s3r.Bucket(self.bucket)
    622     key_list = []
    623     if self.prefix:

File /databricks/python/lib/python3.9/site-packages/boto3/resources/factory.py:527, in ResourceFactory._create_class_partial.<locals>.create_resource(self, *args, **kwargs)
    524     for identifier, value in build_identifiers(identifiers, self):
    525         positional_args.append(value)
--> 527 return partial(
    528     resource_cls, *positional_args, client=self.meta.client
    529 )(*args, **kwargs)

File /databricks/python/lib/python3.9/site-packages/boto3/resources/base.py:125, in ServiceResource.__init__(self, *args, **kwargs)
    123 for identifier in self.meta.identifiers:
    124     if getattr(self, identifier) is None:
--> 125         raise ValueError(f'Required parameter {identifier} not set')

ValueError: Required parameter name not set

Could you please help with this?

Thanks & Regards,
Shubham

[PMRocha]

Hello @jaina15 ,

Currently the dq_specs supports file system and s3 as backends and by default the store_backend for the data quality is set so s3, in order to change this you need to pass it as file_system.
Your dq_specs would look something like this:

"dq_specs": [
        {
            "spec_id": "check_brz_transformation",
            "input_id": "brz_transformation",
            "dq_type": "validator",
            "store_backend": "file_system",
            "local_fs_root_dir": "your_local_path",
            "result_sink_db_table": "cat.test.dq_checks",
            "fail_on_error": False,
            "dq_functions": [
                {
                    "function": "expect_column_values_to_not_be_null",
                    "args": {
                        "column": "name"
                    }
                },
                {
                    "function": "expect_column_values_to_be_between",
                    "args": {
                        "column": "age",
                        "min_value": 18,
                        "max_value": 32
                    },
                }
            ],
        },
    ],

[jaina15]

Hi @PMRocha ,

I updated the my acon as below, now it's not throwing any error but even after dq checks are there it's still inserting those malformed records into the silver table. Also, in the transformation I have added configuration to add a new column of date type, this is also not getting generated. Below is my spec and the outputs:

acon = {
    "input_specs": [
        {
            "spec_id": "brz_test",
            "read_type": "batch",
            "data_format": "delta",
            "db_table": "cat.test.brz_tab",
        }
    ],
    "transform_specs": [
        {
            "spec_id": "brz_transformation",
            "input_id": "brz_test",
            "transformers": [
                {
                    "function": "add_current_date",
                    "args": {
                        "output_col": "lh_created_on"
                    }
                },
                {
                    "function": "with_expressions",
                    "args": {
                        "cols_and_exprs": {
                            "bonus": "age*10",
                            "designation": "CASE WHEN age <= 22 THEN 'Junior ENG' "
                                        "WHEN age>22 and age <=27 THEN 'Senior ENG'"
                                        "ELSE 'Staff ENG' END"
                        }
                    }
                }
            ],
        }
    ],
    "dq_specs": [
        {
            "spec_id": "check_brz_transformation",
            "input_id": "brz_transformation",
            "dq_type": "validator",
            "store_backend": "file_system",
            "local_fs_root_dir": "/FileStore/sj_test/test_dq_checks/",
            "result_sink_db_table": "cat.test.dq_checks",
            "fail_on_error": False,
            "dq_functions": [
                {
                    "function": "expect_column_values_to_not_be_null",
                    "args": {
                        "column": "name"
                    }
                },
                {
                    "function": "expect_column_values_to_be_between",
                    "args": {
                        "column": "age",
                        "min_value": 18,
                        "max_value": 32
                    },
                }
            ],
        },
    ],
    "output_specs": [
        {
            "spec_id": "slv_test",
            "input_id": "check_brz_transformation",
            "data_format": "delta",
            "write_type": "merge",
            "merge_opts": {
                "merge_predicate": """
                    new.name = current.name
                """,
            },
            "db_table": "cat.test.slv_tab"
        }
    ]
}

This is my bronze table:

This is what I am receiving in the silver table. As, per the dq checks the highlighted records should not come under silver right? Also, I am not able to see the lh_created_on column that I am trying to add via transformer.

@jmcorreia

[PMRocha]

Hello @jaina15 ,

the dq process does not filter lines to be input into silver.
The behaviour is when "fail_on_error" is set to True, the process fails when data quality errors are detected. On the other hand, when "fail_on_error" is set to False the process does not fail when data quality errors are detected.

Regarding the missing column, can you try and change your output to console? To check if the column is indeed missing or is being ignored when inserting into the final table.

"output_specs": [
        {
            "spec_id": "slv_test",
            "input_id": "check_brz_transformation",
            "data_format": "console",
        }
    ]

[jaina15]

Hi @PMRocha ,

I am able to see the column in the console output as below:

Any reason, why it's not being added to the final table?
Also, if DQ is not removing or filtering the malformed records then should we go with filter transformers and insert those malformed records separately?

[PMRocha]

Hello @jaina15 ,
was the table created before running this process?
You can enable the automatic schema update by setting spark.databricks.delta.schema.autoMerge.enabled to true in your environment. To do this, you can add exec_env at the end of your acon:

    "exec_env": {
        "spark.databricks.delta.schema.autoMerge.enabled": True,
    },

Regarding the dq errors, it really depends on your use case. You can filter these lines with no name, or fail the process if it is critical.

[jaina15]

This worked thanks!!
I will check what I can try for DQ, as per our requirement we want to move malformed records to another table and only move correct and valid records to the silver layer.