Possibly the fastest DataFrame-agnostic quality check library in town.
cuallee
Meaning good in Aztec (Nahuatl), pronounced: QUAL-E_
This library provides an intuitive API to describe data quality checks initially just for PySpark dataframes v3.3.0. And extended to pandas, snowpark, duckdb, daft and more. It is a replacement written in pure python of the pydeequ framework.
I gave up in deequ as after extensive use, the API is not user-friendly, the Python Callback servers produce additional costs in our compute clusters, and the lack of support to the newest version of PySpark.
As result cuallee was born
This implementation goes in hand with the latest API from PySpark and uses the Observation API to collect metrics at the lower cost of computation. When benchmarking against pydeequ, cuallee uses circa <3k java classes underneath and remarkably less memory.
Support
cuallee is the data quality framework truly dataframe agnostic.
Provider | API | Versions ------- | ----------- | ------ |
snowpark | 1.11.1, 1.4.0 |
pyspark & spark-connect | 4.0.1,3.5.x, 3.4.0, 3.3.x, 3.2.x
| bigquery | 3.4.1 |
pandas| 2.0.2, 1.5.x, 1.4.x
|duckdb | 1.4.0,, 1.0.0,0.10.2,0.9.20.8.0|
polars| 1.34.0, , 1.0.00.19.6
|daft| 0.2.24, 0.2.19
Logos are trademarks of their own brands.
Install
pip install cuallee
Checks
The most common checks for data integrity validations are completeness and uniqueness an example of this dimensions shown below:
from cuallee import Check, CheckLevel # WARN:0, ERR: 1
Nulls on column Id
check = Check(CheckLevel.WARNING, "Completeness")
(
check
.is_complete("id")
.is_unique("id")
.validate(df)
).show() # Returns a pyspark.sql.DataFrame
>[!IMPORTANT]
A new version of the validate output is currently under construction.
Dates
Perhaps one of the most useful features of cuallee is its extensive number of checks for Date and Timestamp values. Including, validation of ranges, set operations like inclusion, or even a verification that confirms continuity on dates using the is_daily check function.
# Unique values on id
check = Check(CheckLevel.WARNING, "CheckIsBetweenDates")
df = spark.sql(
"""
SELECT
explode(
sequence(
to_date('2022-01-01'),
to_date('2022-01-10'),
interval 1 day)) as date
""")
assert (
check.is_between("date", ("2022-01-01", "2022-01-10"))
.validate(df)
.first()
.status == "PASS"
)
Membership
Other common test is the validation of list of values as part of the multiple integrity checks required for better quality data.
df = spark.createDataFrame([[1, 10], [2, 15], [3, 17]], ["ID", "value"])
check = Check(CheckLevel.WARNING, "iscontainedinnumbertest")
check.iscontainedin("value", (10, 15, 20, 25)).validate(df)
Regular Expressions
When it comes to the flexibility of matching, regular expressions are always to the rescue. cuallee makes use of the regular expressions to validate that fields of type String conform to specific patterns.
df = spark.createDataFrame([[1, "isblue"], [2, "hashat"], [3, "is_smart"]], ["ID", "desc"])
check = Check(CheckLevel.WARNING, "haspatterntest")
check.haspattern("desc", r"^is.*t$") # only match issmart 33% of rows.
check.validate(df).first().status == "FAIL"
Anomalies
Statistical tests are a great aid for verifying anomalies on data. Here an example that shows that will PASS only when 40% of data is inside the interquartile range
df = spark.range(10)
check = Check(CheckLevel.WARNING, "IQR_Test")
check.isinsideinterquartile_range("id", pct=0.4)
check.validate(df).first().status == "PASS"
+---+-------------------+-----+-------+------+-----------------------------+-----+----+----------+---------+--------------+------+ |id |timestamp |check|level |column|rule |value|rows|violations|passrate|passthreshold|status| +---+-------------------+-----+-------+------+-----------------------------+-----+----+----------+---------+--------------+------+ |1 |2022-10-19 00:09:39|IQR |WARNING|id |isinsideinterquartile_range|10000|10 |4 |0.6 |0.4 |PASS | +---+-------------------+-----+-------+------+-----------------------------+-----+----+----------+---------+--------------+------+
Workflows (Process Mining)
Besides the commoncitizen-like checks, cuallee offers out-of-the-box real-life checks. For example, suppose that you are working SalesForce or SAP environment. Very likely your business processes will be driven by a lifecycle:
Order-To-CashRequest-To-PayInventory-Logistics-Delivery- Others.
cuallee offers the ability that the sequence of events registered over time, are according to a sequence of events, like the example below:
import pyspark.sql.functions as F
from cuallee import Check, CheckLevel
data = pd.DataFrame({ "name":["herminio", "herminio", "virginie", "virginie"], "event":["new","active", "new", "active"], "date": ["2022-01-01", "2022-01-02", "2022-01-03", "2022-02-04"]} ) df = spark.createDataFrame(data).withColumn("date", F.to_date("date"))
Cuallee Process Mining
Testing that all edges on workflows
check = Check(CheckLevel.WARNING, "WorkflowViolations")
Validate that 50% of data goes from new => active
check.has_workflow("name", "event", "date", [("new", "active")], pct=0.5)
check.validate(df).show(truncate=False)
+---+-------------------+------------------+-------+-------------------------+------------+--------------------+----+----------+---------+--------------+------+ |id |timestamp |check |level |column |rule |value |rows|violations|passrate|passthreshold|status| +---+-------------------+------------------+-------+-------------------------+------------+--------------------+----+----------+---------+--------------+------+ |1 |2022-11-07 23:08:50|WorkflowViolations|WARNING|('name', 'event', 'date')|has_workflow|(('new', 'active'),)|4 |2.0 |0.5 |0.5 |PASS | +---+-------------------+------------------+-------+-------------------------+------------+--------------------+----+----------+---------+--------------+------+
Assertions
[2024-09-28] โจ New feature! Return a simple true|false as a unified result for your check import pandas as pd from cuallee import Check df = pd.DataFrame({"X":[1,2,3]}) .ok(dataframe) method of a check will call validate and then verify that all rules are PASS
assert Check().is_complete("X").ok(df)
Controls
Simplify the entire validation of a dataframe in a particular dimension.import pandas as pd
from cuallee import Control
df = pd.DataFrame({"X":[1,2,3], "Y": [10,20,30]})
Checks all columns in dataframe for using is_complete check
Control.completeness(df)
cuallee VS pydeequ
In the test folder there are docker containers with the requirements to match the tests. Also a perftest.py available at the root folder for interests.
# 1000 rules / # of seconds
cuallee: โโโโโโโโโโโโโโโโโโโโโโโโโ 162.00 pydeequ: โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ 322.00
Catalogue
Check | Description | DataType ------- | ----------- | ---- iscomplete | Zero nulls | agnostic_ isunique | Zero duplicates | agnostic_ isprimarykey | Zero duplicates | agnostic arecomplete | Zero nulls on group of columns | agnostic_ areunique | Composite primary key check | agnostic_ iscompositekey | Zero duplicates on multiple columns | agnostic isgreaterthan | col > x | numeric ispositive | col > 0 | numeric_ isnegative | col < 0 | numeric_ isgreaterorequalthan | col >= x | numeric islessthan | col < x | numeric islessorequalthan | col <= x | numeric isequalthan | col == x | numeric iscontainedin | col in [a, b, c, ...] | agnostic isin | Alias of iscontainedin | agnostic_ notcontainedin | col not in [a, b, c, ...] | agnostic notin | Alias of notcontainedin | agnostic_ isbetween | a <= col <= b | numeric, date_ haspattern | Matching a pattern defined as a regex | string_ islegit | String not null & not empty ^\S$ | string_ hasmin | min(col) == x | numeric_ hasmax | max(col) == x | numeric_ hasstd | ฯ(col) == x | numeric_ hasmean | ฮผ(col) == x | numeric_ hassum | ฮฃ(col) == x | numeric_ haspercentile | %(col) == x | numeric_ hascardinality | count(distinct(col)) == x | agnostic_ hasinfogain | count(distinct(col)) > 1 | agnostic_ hasmaxby | A utilitary predicate for max(cola) == x for max(colb) | agnostic hasminby | A utilitary predicate for min(cola) == x for min(colb) | agnostic hascorrelation | Finds correlation between 0..1 on corr(cola, colb) | numeric_ hasentropy | Calculates the entropy of a column entropy(col) == x for classification problems | numeric_ isinsideinterquartilerange | Verifies column values reside inside limits of interquartile range Q1 <= col <= Q3 used on anomalies. | numeric_ isinmillions | col >= 1e6 | numeric isinbillions | col >= 1e9 | numeric istminus1 | For date fields confirms 1 day ago t-1 | date_ istminus2 | For date fields confirms 2 days ago t-2 | date_ istminus3 | For date fields confirms 3 days ago t-3 | date_ istminusn | For date fields confirms n days ago t-n | date_ istoday | For date fields confirms day is current date t-0 | date_ isyesterday | For date fields confirms 1 day ago t-1 | date_ isonweekday | For date fields confirms day is between Mon-Fri | date isonweekend | For date fields confirms day is between Sat-Sun | date isonmonday | For date fields confirms day is Mon | date isontuesday | For date fields confirms day is Tue | date isonwednesday | For date fields confirms day is Wed | date isonthursday | For date fields confirms day is Thu | date isonfriday | For date fields confirms day is Fri | date isonsaturday | For date fields confirms day is Sat | date isonsunday | For date fields confirms day is Sun | date isonschedule | For date fields confirms time windows i.e. 9:00 - 17:00 | timestamp isdaily | Can verify daily continuity on date fields by default. [2,3,4,5,6] which represents Mon-Fri in PySpark. However new schedules can be used for custom date continuity | date_ hasworkflow | Adjacency matrix validation on 3-column graph, based on group, event, order columns. | agnostic_ iscustom | User-defined custom function applied to dataframe for row-based validation. | agnostic_ satisfies | An open SQL expression builder to construct custom checks | agnostic validate | The ultimate transformation of a check with a dataframe input for validation | agnostic
Controls pyspark
Check | Description | DataType ------- | ----------- | ---- completeness | Zero nulls | agnostic information | Zero nulls and cardinality > 1 | agnostic intelligence | Zero nulls, zero empty strings and cardinality > 1 | agnostic percentagefill | % rows not empty | agnostic_ percentageempty | % rows empty | agnostic_
ISO Standard
A new module has been incorporated incuallee==0.4.0 which allows the verification of International Standard Organization columns in data frames. Simply access the check.iso interface to add the set of checks as shown below.
Check | Description | DataType ------- | ----------- | ---- iso4217 | currency compliant ccy | string_ iso3166 | country compliant country | string_
df = spark.createDataFrame([[1, "USD"], [2, "MXN"], [3, "CAD"], [4, "EUR"], [5, "CHF"]], ["id", "ccy"])
check = Check(CheckLevel.WARNING, "ISO Compliant")
check.iso.iso_4217("ccy")
check.validate(df).show()
+---+-------------------+-------------+-------+------+---------------+--------------------+----+----------+---------+--------------+------+
| id| timestamp| check| level|column| rule| value|rows|violations|passrate|passthreshold|status|
+---+-------------------+-------------+-------+------+---------------+--------------------+----+----------+---------+--------------+------+
| 1|2023-05-14 18:28:02|ISO Compliant|WARNING| ccy|iscontainedin|{'BHD', 'CRC', 'M...| 5| 0.0| 1.0| 1.0| PASS|
+---+-------------------+-------------+-------+------+---------------+--------------------+----+----------+---------+--------------+------+
Snowflake Connection
In order to establish a connection to your SnowFlake accountcuallee relies in the following environment variables to be avaialble in your environment:
SF_ACCOUNTSF_USERSF_PASSWORDSF_ROLESF_WAREHOUSESF_DATABASESF_SCHEMA
Spark Connect
Just add the environment variableSPARK_REMOTE to your remote session, then cuallee will connect using
sparkconnect = SparkSession.builder.remote(os.getenv("SPARKREMOTE")).getOrCreate()
and convert all checks to select as opposed to Observation API compute instructions.
Databricks Connection
By defaultcuallee will search for a SparkSession available in the globals so there is literally no need to SparkSession.builderDuckDB
For testing on duckdb simply pass your table name to your check et voilร
import duckdb
conn = duckdb.connect(":memory:")
check = Check(CheckLevel.WARNING, "DuckDB", table_name="temp/taxi/*.parquet")
check.is_complete("VendorID")
check.iscomplete("tpeppickup_datetime")
check.validate(conn)
id timestamp check level column rule value rows violations passrate passthreshold status 0 1 2022-10-31 23:15:06 test WARNING VendorID is_complete N/A 19817583 0.0 1.0 1.0 PASS 1 2 2022-10-31 23:15:06 test WARNING tpeppickupdatetime is_complete N/A 19817583 0.0 1.0 1.0 PASS
Roadmap
100% data frame agnostic implementation of data quality checks. Define once, run everywhere
[x] PySpark 3.5.0[x] PySpark 3.4.0[x] PySpark 3.3.0[x] PySpark 3.2.x[x] Snowpark DataFrame[x] Pandas DataFrame[x] DuckDB Tables[x] BigQuery Client[x] Polars DataFrame[*] Dagster Integration[x] Spark Connect[x] Daft- [-] PDF Report
- [ ] Metadata check
- [ ] Help us in a discussion?
Whilst expanding the functionality feels a bit as an overkill because you most likely can connect spark via its drivers to whatever DBMS of your choice. In the desire to make it even more user-friendly we are aiming to make cuallee portable to all the providers above.
Authors
- canimus / Herminio Vazquez / ๐ฒ๐ฝ
- vestalisvirginis / Virginie Grosboillot / ๐ซ๐ท
Contributors
Guidelines
Documentation
Paper
cuallee has been published in the Journal of Open Source Software
Vazquez et al., (2024). cuallee: A Python package for data quality checks across multiple DataFrame APIs. Journal of Open Source Software, 9(98), 6684, https://doi.org/10.21105/joss.06684 If you use cuallee please consider citing this work. Citation
License
Apache License 2.0 Free for commercial use, modification, distribution, patent use, private use. Just preserve the copyright and license.Made with โค๏ธ in Utrecht ๐ณ๐ฑ
Maintained over โ from Ljubljana ๐ธ๐ฎ
Extended ๐ by contributions all over the ๐