Metadata-Version: 2.4
Name: spark-data-quality
Version: 1.0.8
Summary: SparkDQAgent — Data Quality validation package for K8s Spark pods
Author-email: khailas <khailas.rangath@saal.ai>
License-Expression: MIT
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: OS Independent
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.28.0
Requires-Dist: great-expectations==0.18.12
Requires-Dist: trino>=0.320.0
Requires-Dist: PyYAML>=6.0
Provides-Extra: spark
Requires-Dist: pyspark>=3.1.1; extra == "spark"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Dynamic: license-file

# spark-data-quality

Data Quality validation library for Kubernetes Spark pods. Runs Great Expectations
and Trino SQL checks against any table, then persists results to the Data Quality
Engine API.



## Usage — By Table Name

When you only have the table's fully qualified name (`catalog.schema.table`),
no `table_id` or `test_suite_id` is needed. The DQ Engine auto-resolves all
IDs from Discovery Hub.

```python
from spark_dq.quality import SparkDQAgent

agent = SparkDQAgent(
    catalog="my_catalog",
    schema="my_schema",
    table="my_table",
    data_quality_url="http://dq-engine:8000/api/v1/spark",
    trino_host="trino:443",
    trino_user="user",
    trino_pwd="pwd",
)

cfg = agent.fetch_table_config()
results = agent.execute_data_quality(df)
```

### `fetch_table_config()`

Calls `GET /config` on the DQ Engine with `catalog_name`, `schema_name`, and
`table_name` as query parameters. The DQ Engine then calls Discovery Hub's
`GET /ds/table/lookup/` to resolve the internal `table_id`, `schema_id`, and
`catalog_id`. Once the table is identified, the DQ Engine fetches all active
test suites and their expectations for that table and returns them as a config
dict containing:

- `quality_query` — the SQL query to load table data
- `table_type` — `TABLE` or `VIRTUAL_VIEW`
- `catalog_type` — `managed` or `unmanaged`
- `scan_limit` — max rows to scan (if configured)
- `suites` — list of test suites, each with its expectations
- `table_id`, `schema_id`, `catalog_id` — resolved IDs

The config is cached after the first call, so subsequent calls return instantly.

### `execute_data_quality(df)`

Takes a Spark DataFrame and runs all suites from the config against it.

1. Reads the config (calls `fetch_table_config()` internally if not already cached)
2. Backfills `table_id` from the config if it was not provided at init
3. For each suite, splits expectations into two concurrent paths:
   - **Trino fast-path** — null checks, uniqueness, range, regex (single SQL query)
   - **GE slow-path** — all other expectation types (parallel Great Expectations validators)
4. Saves results to the DQ Engine via `POST /save-results`
5. Returns per-suite statistics (evaluated, passed, failed, success %)

Suites run in parallel (up to 8 concurrently). If Trino credentials are not
provided, all expectations run through Great Expectations only.

## License

MIT
