Data Quality checks failing with weird error

[divyadarshan01]

I am trying a very basic data transformation and data quality checks to review the framework. However while running the quality checks, it is failing with unexpected error. Can someone please guide me what could be wrong. I am trying the setup in WSL2 environment and dont have S3 support.

NOTE: I had to downgrade the python dependencies to 3.9 compatibility as that is environment available in databricks. But still using

great-expectations==0.18.8

Code is working fine overall (hence no python dependency issues) but could not debug this error.

My configuration is:

def agg_query(df: DataFrame) -> DataFrame:
    session = ExecEnv.SESSION
    df.createOrReplaceTempView("add_region_column")
    
    kpi_df = session.sql("""SELECT 
                                    MANDT,
                                    BUZEI,
                                    BELNR,
                                    SUM(BUKRS) AS SUM_SALES 
                                FROM 
                                    add_region_column 
                                GROUP BY 
                                    MANDT,
                                    BUZEI,
                                    BELNR    
                                """
                            )
    return kpi_df
acon = {
  "algorithm": "data_load",
  "input_specs": [
    {
      "spec_id": "input_bseg",
      "read_type": "batch",
      "data_format": "csv",
      "options": {
        "header": "true",
        "delimiter": ";",
        "inferSchema": "true"
      },
      "location": "file:///home/user/dev/smallbseg.csv",
      "temp_view": "bseg"
    }
  ],
  "transform_specs": [
    {
      "spec_id": "filter_recent_date",
      "input_id": "input_bseg",
      "transformers": [
        {
          "function": "expression_filter",
          "args": {
            "exp": "to_date(augdt, 'yyyyMMdd') >= to_date('20210218', 'yyyyMMdd') AND augdt != '00000000'"
          }
        }
      ]
    },
    {
      "spec_id": "add_region_column",
      "temp_view": "add_region_column",
      "input_id": "filter_recent_date",
      "transformers": [
        {
          "function": "with_expressions",
          "args": {
            "cols_and_exprs": 
            {
            "region": "CASE WHEN BUZID = 'W' THEN 'North America' WHEN BUZID = 'M' THEN 'ASIA' ELSE 'Other' END"
          }
          }
        }
      ]
    },
    {
      "spec_id": "aggregate_sales",
      "input_id": "add_region_column",
      "transformers": [
        {
          "function": "custom_transformation",
          "args": {
            "custom_transformer": agg_query
          }
        }
      ]
    }
  ],
  "dq_specs": [
    {
      "spec_id": "validate_bseg_data",
      "input_id": "aggregate_sales",
      "dq_type": "validator",
      "store_backend": "file_system",
      "local_fs_root_dir": "/home/user/dev/dq_v1/",
      "tag_source_data": False,
      "unexpected_rows_pk": "BELNR",
      "fail_on_error": False,
      "dq_functions": [
        {
          "function": "expect_column_values_to_not_be_null_or_empty_string",
          "args": {
            "column": "BELNR"
          }
        },
        {
          "function": "expect_column_values_to_be_in_set",
          "args": {
            "column": "MANDT",
            "value_set": ["840"]
          }
        }
      ],
      "critical_functions": [
        {
          "function": "expect_table_column_count_to_be_between",
          "args": {
            "max_value": 10
          }
        }
      ]
    }
  ],
  "output_specs": [
    {
      "spec_id": "sales_processed",
      "input_id": "aggregate_sales",
      "data_format": "console"
    }
  ]
}

dummy_deliveries_transformed = load_data(acon=acon)

My csv looks like:

MANDT;BUKRS;BELNR;GJAHR;BUZEI;BUZID;AUGDT
840;1000;2000038494;2021;001;M;00000000
840;1000;2000038494;2021;002;W;20210223
840;1000;2000038494;2021;003;M;00000000
840;1000;2000038494;2021;004;W;20210223
840;1000;2000038494;2021;005;M;00000000
840;1000;2000038494;2021;006;W;20210223
840;1000;2000038494;2021;007;M;00000000
840;1000;2000038494;2021;008;W;20210223
840;1000;2000038495;2021;001;M;00000000
840;1000;2000038495;2021;002;W;20210218
840;1000;2000038495;2021;003;M;00000000
840;1000;2000038495;2021;004;W;20210218
840;1000;2000038495;2021;005;M;00000000
840;1000;2000038495;2021;006;W;20210218
840;1000;2000038495;2021;007;M;00000000
840;1000;2000038495;2021;008;W;20210218
840;1000;3500037862;2021;001;;20210214
840;1000;3500037862;2021;002;;00000000
840;1000;3500037862;2021;003;;20210214

When I am running the library, getting below issue:

  "expectation_type": "expect_column_values_to_not_be_null_or_empty_string",
  "kwargs": {
    "column": "BELNR",
    "batch_id": "aa1fec2cde52de0999d6b145654ffc50"
  },
  "meta": {}
}
                            raised the following exception:
                            Error: The unexpected_index_column 'B' does not exist in Spark DataFrame. Please check your configuration and try again.
2025-03-01 12:59:37,856 — lakehouse_engine.dq_processors.dq_factory — ERROR — The expectation {
  "expectation_type": "expect_column_values_to_be_in_set",
  "kwargs": {
    "column": "MANDT",
    "value_set": [
      "840"
    ],
    "batch_id": "aa1fec2cde52de0999d6b145654ffc50"
  },
  "meta": {}
}
                            raised the following exception:
                            Error: The unexpected_index_column 'B' does not exist in Spark DataFrame. Please check your configuration and try again```

[epilif1017a]

Hi :)

I tested using the containerised environment in 1.24.0 and was not able to reproduce. Here is the result log (it worked as expected):

result.log

In regards to running it on a python 3.9 environment, I don't think we support that anymore, as I tried to build the requirements lockfiles again and rebuild the docker image using python 3.9 and was not able to install the lakehouse_engine due to not matching requirements. Databricks runtime is using Python 3.10 for a while now https://docs.databricks.com/aws/en/release-notes/runtime/13.3lts, so maybe you want to at least use that? we add used 3.10 for a long time but not sure if our current requirements still support that as we have been building and testing on 3.11 for a while now.

Aren't you able to run 3.11 on databricks? or at least 3.10?

[divyadarshan01]

But result.log shows the same error I am getting.

We have DBR 12.2 only as of now hence my constraint of python 3.9. I am requesting DBR 15.4 so hopefully will have support for python 3.11 in near time.

But again coming back to the issue, I can see two validations being failed

Note: Replying from mobile so may have overseen something so feel free to correct me.

Thanks

[divyadarshan01]

Found the issue. It was due to "unexpected_rows_pk" key. I was passing single column which was not unique across my dataset. However when I passed the combination of two columns, everything worked fine.
"unexpected_rows_pk": ["BELNR","BUZEI"],

Error throwed me off and I was focused on finding issue in dq_functions.
Btw I tested the solution with both python 3.12 and my degraded version of 3.9. In both cases, it is working fine.

Thanks a lot.

[epilif1017a]

Hmmm to be honest I did not even noticed that. But that's a super strange error that does not resonate at all with a non unique pk.

@jmcorreia @PMRocha might be worth for us to check what's going on here.

[jmcorreia]

Hi @divyadarshan01 ,
great that you found the answer :)
That makes sense as it is expecting a list as you can see here https://adidas.github.io/lakehouse-engine-docs/reference/packages/core/definitions.html#packages.core.definitions.DQSpec.
In any case, I just created an internal issue, for us to have a look and check if we can provide a better logging message or validation.
Thanks.