PyDeequ

https://pydeequ.readthedocs.io/en/latest/
Admin
PyDeequ is a Python library for data quality validation and testing, enabling users to build and run...
Tokens:14,148
Snippets:78
Trust Score:8.3
Update:1 year ago
Show doc for...
Tokens:
Raw
### Install PyDeequ using pip

Source: https://pydeequ.readthedocs.io/en/latest/README

Installs the PyDeequ library and its dependencies using pip. Ensure you have Python and pip installed.

```bash
pip install pydeequ
```

--------------------------------

### Installing SDKMAN: Version Manager for SDKs

Source: https://pydeequ.readthedocs.io/en/latest/README

This code installs SDKMAN, a tool for managing multiple Software Development Kit versions on Unix-based systems. It provides a command-line interface for installation and management. The process involves downloading a script, initializing SDKMAN, and verifying the installation.

```bash
$ curl -s https://get.sdkman.io | bash
If the environment needs tweaking for SDKMAN to be installed,
the installer will prompt you accordingly and ask you to restart.

Next, open a new terminal or enter:

$ source "$HOME/.sdkman/bin/sdkman-init.sh"

Lastly, run the following code snippet to ensure that installation succeeded:

$ sdk version
```

--------------------------------

### Installing Apache Spark using SDKMAN

Source: https://pydeequ.readthedocs.io/en/latest/README

This code snippet shows how to install Apache Spark using SDKMAN. It includes commands to list available Spark versions and install a specific version, such as Spark 3.0.2. This is a prerequisite for running distributed data processing jobs with PyDeequ.

```bash
List the Apache Spark versions:
$ sdk list spark

To install For Spark 3
$ sdk install spark 3.0.2
```

--------------------------------

### Installing Java using SDKMAN

Source: https://pydeequ.readthedocs.io/en/latest/README

This code demonstrates how to install specific Java versions using the SDKMAN tool. It includes commands to list available Java versions and install particular versions, such as Java 11 and Java 8. This is essential for ensuring compatibility with PyDeequ and other Java-based tools.

```bash
List the AdoptOpenJDK OpenJDK versions
$ sdk list java

To install For Java 11
$ sdk install java 11.0.10.hs-adpt

To install For Java 11
$ sdk install java 8.0.292.hs-adpt
```

--------------------------------

### Execute PyDeequ Analyses with AnalysisRunner

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Illustrates the use of AnalysisRunner to initiate an analysis run on a DataFrame. It acts as a starting point for constructing an AnalysisRun.

```python
from pydeequ.analyzers import AnalysisRunner

# Assuming spark is a SparkSession and df is a DataFrame
analysisRunner = AnalysisRunner(spark)
analysisRunBuilder = analysisRunner.onData(df)

# Further steps would involve adding analyzers to analysisRunBuilder and running it.
```

--------------------------------

### Initialize Spark Session with PyDeequ

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Initialize a Spark session and configure it to include PyDeequ dependencies. This setup is necessary before running any PyDeequ operations.

```python
from pyspark.sql import SparkSession, Row
import pydeequ

# Initialize Spark session with PyDeequ dependencies
spark = (SparkSession
    .builder
    .config("spark.jars.packages", pydeequ.deequ_maven_coord)
    .config("spark.jars.excludes", pydeequ.f2j_maven_coord)
    .getOrCreate())

# Create sample DataFrame
df = spark.sparkContext.parallelize([
    Row(a="foo", b=1, c=5),
    Row(a="bar", b=2, c=6),
    Row(a="baz", b=3, c=None)
]).toDF()
```

--------------------------------

### Managing Python Dependencies with Poetry

Source: https://pydeequ.readthedocs.io/en/latest/README

This code provides common Poetry commands for managing Python project dependencies. It includes commands for installing dependencies, updating them, and displaying dependency information, including a tree view and outdated packages. Poetry is used to handle PyDeequ's project dependencies.

```bash
poetry install

poetry update

# --tree: List the dependencies as a tree.
# --latest (-l): Show the latest version.
# --outdated (-o): Show the latest version but only for packages that are outdated.
poetry show -o
```

--------------------------------

### Load Metrics from Repository with PyDeequ

Source: https://pydeequ.readthedocs.io/en/latest/README

Loads previously saved metrics from a file-based metrics repository. This example shows filtering results before a specific time and retrieving them as a DataFrame for a specified analyzer.

```python
result_metrep_df = repository.load() \
    .before(ResultKey.current_milli_time()) \
    .forAnalyzers([ApproxCountDistinct('b')]) \
    .getSuccessMetricsAsDataFrame()

```

--------------------------------

### Get PyDeequ Analysis Results as JSON

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Demonstrates how to get the analysis run results in JSON format. Similar to the DataFrame method, it allows filtering for specific analyzers.

```python
from pydeequ.analyzers import AnalyzerContext

# Assuming spark is a SparkSession and analyzerContext is an AnalyzerContext object

# Get all success metrics as JSON
json_results = AnalyzerContext.successMetricsAsJson(spark, analyzerContext)

# Get success metrics for specific analyzers as JSON
# specific_analyzers = [...] # List of analyzer objects
# json_specific_results = AnalyzerContext.successMetricsAsJson(spark, analyzerContext, forAnalyzers=specific_analyzers)
```

--------------------------------

### Get PyDeequ Analysis Results as DataFrame

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Shows how to retrieve the results of an analysis run as a Spark DataFrame. This method can optionally filter results for specific analyzers.

```python
from pydeequ.analyzers import AnalyzerContext

# Assuming spark is a SparkSession and analyzerContext is an AnalyzerContext object

# Get all success metrics as a DataFrame
df_results = AnalyzerContext.successMetricsAsDataFrame(spark, analyzerContext)

# Get success metrics for specific analyzers
# specific_analyzers = [...] # List of analyzer objects
# df_specific_results = AnalyzerContext.successMetricsAsDataFrame(spark, analyzerContext, forAnalyzers=specific_analyzers)
```

--------------------------------

### Set up a PySpark Session with PyDeequ

Source: https://pydeequ.readthedocs.io/en/latest/README

Initializes a PySpark session configured to use PyDeequ. This involves specifying Spark JAR packages for Deequ and excluding FastJava dependency. It also creates a sample Spark DataFrame.

```python
from pyspark.sql import SparkSession, Row
import pydeequ

spark = (SparkSession
    .builder
    .config("spark.jars.packages", pydeequ.deequ_maven_coord)
    .config("spark.jars.excludes", pydeequ.f2j_maven_coord)
    .getOrCreate())

df = spark.sparkContext.parallelize([
            Row(a="foo", b=1, c=5),
            Row(a="bar", b=2, c=6),
            Row(a="baz", b=3, c=None)]).toDF()

```

--------------------------------

### Run PyDeequ Analysis with AnalysisRunBuilder

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Demonstrates how to use AnalysisRunBuilder to add analyzers and execute an analysis. This class allows for chaining method calls for a fluent API.

```python
from pydeequ.analyzers import AnalysisRunBuilder
from pydeequ.analyzers import ApproxCountDistinct
from pydeequ.metrics import ResultKey

# Assuming spark is a SparkSession and df is a DataFrame
runBuilder = AnalysisRunBuilder(spark, df)

# Add an analyzer
analyzer = ApproxCountDistinct("my_column")
runBuilder.addAnalyzer(analyzer)

# Run the analysis
runBuilder.run()

# Save or append results
resultKey = ResultKey("my_run_id")
runBuilder.saveOrAppendResult(resultKey)

# Use a metrics repository
# from pydeequ.repository import MetricsRepository
# repository = MetricsRepository.create(spark, "path/to/repo")
# runBuilder.useRepository(repository)
```

--------------------------------

### Use InMemoryMetricsRepository for Temporary Storage

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Illustrates the usage of InMemoryMetricsRepository for testing purposes or when temporary storage of metrics is sufficient. It shows how to initialize, run analyses, and immediately load metrics from memory.

```python
from pydeequ.repository import *
from pydeequ.analyzers import *

# Use in-memory repository for testing or temporary storage
repository = InMemoryMetricsRepository(spark)
resultKey = ResultKey(spark, ResultKey.current_milli_time(), {'env': 'test'})

analysisResult = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(Size()) \
    .addAnalyzer(Completeness('a')) \
    .useRepository(repository) \
    .saveOrAppendResult(resultKey) \
    .run()

# Load immediately from memory
metrics_df = repository.load().getSuccessMetricsAsDataFrame()
```

--------------------------------

### PyDeequ Verification API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Documentation related to the verification module in PyDeequ.

```APIDOC
## PyDeequ Verification API

### Description
Information about the verification functionalities.
```

--------------------------------

### PyDeequ Metrics Repository Classes

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Documentation for the MetricsRepository base class and its implementations, FileSystemMetricsRepository and InMemoryMetricsRepository, used for managing analysis results.

```APIDOC
## Class _pydeequ.repository.FileSystemMetricsRepository

### Description
High level FileSystemMetricsRepository Interface.

### Parameters
* **spark_session** (SparkSession) - SparkSession
* **path** (Optional[str]) - The location of the file metrics repository.

### Bases
- `MetricsRepository`

## Class _pydeequ.repository.InMemoryMetricsRepository

### Description
High level InMemoryMetricsRepository Interface.

### Parameters
* **spark_session** (SparkSession) - SparkSession

### Bases
- `MetricsRepository`

## Class _pydeequ.repository.MetricsRepository

### Description
Base class for Metrics Repository.

### Methods
#### `after(_dateTime: int_)`
Only look at AnalysisResults with a result key with a greater value.
* **Parameters**
  * **dateTime** (int) - The minimum dateTime of AnalysisResults to look at.

#### `before(_dateTime: int_)`
Only look at AnalysisResults with a result key with a smaller value.
* **Parameters**
  * **dateTime** (int) - The maximum dateTime of AnalysisResults to look at.

#### `forAnalyzers(_analyzers: list_)`
Choose all metrics that you want to load.
* **Parameters**
  * **analyzers** (list) - List of analyzers whose resulting metrics you want to load.

#### `getSuccessMetricsAsDataFrame(_withTags: Optional[list] = None_, _pandas: bool = False_)`
Get the AnalysisResult as DataFrame.
* **Parameters**
  * **withTags** (Optional[list]) - List of tags to filter previous Metrics Repository runs with.
  * **pandas** (bool) - Whether to return as a pandas DataFrame.

#### `getSuccessMetricsAsJson(_withTags: Optional[list] = None_)`
Get the AnalysisResult as JSON.
* **Parameters**
  * **withTags** (Optional[list]) - List of tags to filter previous Metrics Repository runs with.

#### `load()`
Get a builder class to construct a loading query to get AnalysisResults.

#### `withTagValues(_tagValues: dict_)`
Filter out results that don’t have specific values for specific tags.
* **Parameters**
  * **tagValues** (dict) - Dict with tag names and the corresponding values to filter for.

### Class Methods
#### `_classmethod _helper_metrics_file(_spark_session: SparkSession_, _filename: str = 'metrics.json'_)`
Helper method to create the metrics file for storage.
* **Parameters**
  * **spark_session** (SparkSession) - Spark session.
  * **filename** (str) - The name of the metrics file.

## Class _pydeequ.repository.ResultKey

### Description
Information that uniquely identifies an AnalysisResult.

### Parameters
* **spark_session** (SparkSession) - SparkSession
* **dataSetDate** (Optional[int]) - Date of the result key.
* **tags** (Optional[dict]) - A map with additional annotations.

### Static Methods
#### `_static _current_milli_time()`
Get current time in milliseconds.
```

--------------------------------

### Save and Load Metrics using FileSystemMetricsRepository

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Demonstrates how to set up a FileSystemMetricsRepository to persist and track data quality metrics over time. It covers saving analysis results with specific keys and timestamps, and loading historical metrics with various filters.

```python
from pydeequ.repository import *
from pydeequ.analyzers import *

# Setup filesystem metrics repository
metrics_file = FileSystemMetricsRepository.helper_metrics_file(spark, 'metrics.json')
repository = FileSystemMetricsRepository(spark, metrics_file)

# Create result key with timestamp and tags
key_tags = {'tag': 'daily_quality_check', 'dataset': 'sales'}
resultKey = ResultKey(spark, ResultKey.current_milli_time(), key_tags)

# Run analysis and save to repository
analysisResult = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(ApproxCountDistinct('b')) \
    .addAnalyzer(Completeness('c')) \
    .addAnalyzer(Mean('b')) \
    .useRepository(repository) \
    .saveOrAppendResult(resultKey) \
    .run()

# Load metrics from repository with filters
result_df = repository.load() \
    .before(ResultKey.current_milli_time()) \
    .after(1609459200000)  # timestamp filter \
    .forAnalyzers([ApproxCountDistinct('b'), Completeness('c')]) \
    .withTagValues({'dataset': 'sales'}) \
    .getSuccessMetricsAsDataFrame()

result_df.show()

# Get results as JSON
result_json = repository.load() \
    .forAnalyzers([Mean('b')]) \
    .getSuccessMetricsAsJson(withTags=['daily_quality_check'])
```

--------------------------------

### Verify Constraints with PyDeequ

Source: https://pydeequ.readthedocs.io/en/latest/README

Performs data validation using `VerificationSuite` by defining a `Check` with various data quality rules (e.g., size, min, completeness, uniqueness). The results are then presented as a DataFrame.

```python
from pydeequ.checks import *
from pydeequ.verification import *

check = Check(spark, CheckLevel.Warning, "Review Check")

checkResult = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(
        check.hasSize(lambda x: x >= 3) \
        .hasMin("b", lambda x: x == 0) \
        .isComplete("c")  \
        .isUnique("a")  \
        .isContainedIn("a", ["foo", "bar", "baz"]) \
        .isNonNegative("b")) \
    .run()

checkResult_df = VerificationResult.checkResultsAsDataFrame(spark, checkResult)
checkResult_df.show()

```

--------------------------------

### Running PyDeequ Tests Locally with Poetry

Source: https://pydeequ.readthedocs.io/en/latest/README

This command runs the local tests for PyDeequ using Poetry. It assumes that tests are located in the `tests/dataquality` and `tests/jobs` directories. Running tests locally is a crucial step in the development workflow to ensure code changes do not introduce regressions.

```bash
$ poetry run pytest
```

--------------------------------

### PyDeequ Scala Utilities API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Utility functions for interacting with Scala from PyDeequ.

```APIDOC
## PyDeequ Scala Utilities API

### Description
Utility functions for seamless integration with Scala.

### Utilities
- `PythonCallback`
- `ScalaFunction1`
- `ScalaFunction2`
- `get_or_else_none()`
- `java_list_to_python_list()`
- `scala_get_default_argument()`
- `scala_map_to_dict()`
- `scala_map_to_java_map()`
- `to_scala_map()`
- `to_scala_seq()`
```

--------------------------------

### Save Metrics to Repository with PyDeequ

Source: https://pydeequ.readthedocs.io/en/latest/README

Demonstrates saving computed metrics to a file-based metrics repository using `FileSystemMetricsRepository`. It configures a `ResultKey` and uses `AnalysisRunner` to save the analysis results.

```python
from pydeequ.repository import *
from pydeequ.analyzers import *

metrics_file = FileSystemMetricsRepository.helper_metrics_file(spark, 'metrics.json')
repository = FileSystemMetricsRepository(spark, metrics_file)
key_tags = {'tag': 'pydeequ hello world'}
resultKey = ResultKey(spark, ResultKey.current_milli_time(), key_tags)

analysisResult = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(ApproxCountDistinct('b')) \
    .useRepository(repository) \
    .saveOrAppendResult(resultKey) \
    .run()

```

--------------------------------

### PyDeequ Repository API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Documentation for the metrics repository components in PyDeequ for storing and retrieving analysis results.

```APIDOC
## PyDeequ Repository API

### Description
Details on components for managing metrics repositories.

### Repository Components
- `FileSystemMetricsRepository`
- `InMemoryMetricsRepository`
- `MetricsRepository`
- `ResultKey`
```

--------------------------------

### FileSystemMetricsRepository Class in pydeequ.repository

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Represents a high-level interface for a file system-based metrics repository. It allows users to interact with stored analysis results, filter them by date and tags, and retrieve them as DataFrames or JSON. Dependencies include SparkSession.

```python
class _pydeequ.repository.FileSystemMetricsRepository(_spark_session : SparkSession_, _path : Optional[str] = None_)
    Bases: MetricsRepository
    High level FileSystemMetricsRepository Interface 

    Parameters:
        
      * **spark_session** – SparkSession
      * **path** – The location of the file metrics repository.
```

--------------------------------

### PyDeequ Checks API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Documentation for the checks framework in PyDeequ for defining and running data quality checks.

```APIDOC
## PyDeequ Checks API

### Description
Provides details on defining and executing data quality checks.

### Check Components
- `Check`
- `CheckLevel`
- `CheckResult`
- `CheckStatus`
- `ConstrainableDataTypes`
```

--------------------------------

### Suggest Constraints using PyDeequ

Source: https://pydeequ.readthedocs.io/en/latest/README

Employs `ConstraintSuggestionRunner` to automatically suggest data quality constraints based on the data in a Spark DataFrame. It uses default rules to generate suggestions in JSON format.

```python
from pydeequ.suggestions import *

suggestionResult = ConstraintSuggestionRunner(spark)
             .onData(df)
             .addConstraintRule(DEFAULT())
             .run()

# Constraint Suggestions in JSON format
print(suggestionResult)

```

--------------------------------

### PyDeequ Profiles API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Details regarding data profiling functionalities in PyDeequ.

```APIDOC
## PyDeequ Profiles API

### Description
Information about data profiling features.
```

--------------------------------

### PyDeequ Suggestions API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Information on suggestions provided by PyDeequ for data quality improvements.

```APIDOC
## PyDeequ Suggestions API

### Description
Details on data quality suggestions generated by PyDeequ.
```

--------------------------------

### Implement Custom Compliance Checks with PyDeequ

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Details how to define and execute custom data quality checks using PyDeequ's Check and VerificationSuite APIs. This includes checks based on SQL predicates and combinations of built-in constraints with hints.

```python
from pydeequ.checks import *
from pydeequ.verification import *

check = Check(spark, CheckLevel.Error, "Business Rules")

# Custom SQL predicate compliance
verification = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(
        check.satisfies(
            "b * 2 < c",
            "Double b is less than c",
            lambda x: x > 0.5
        )
    ) \
    .run()

# Multiple compliance rules
check2 = Check(spark, CheckLevel.Warning, "Data Constraints")
verification2 = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(
        check2
        .hasCompleteness("a", lambda x: x == 1.0, hint="Column a must be complete")
        .hasMin("b", lambda x: x >= 0, hint="b must be non-negative")
        .hasMax("c", lambda x: x <= 100, hint="c must be <= 100")
    ) \
    .run()
```

--------------------------------

### Generate Automatic Constraint Suggestions

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Employ the ConstraintSuggestionRunner to automatically generate data quality constraint suggestions based on the input DataFrame. This helps in discovering potential data quality rules.

```python
from pydeequ.suggestions import *

# Automatically generate constraint suggestions based on data
suggestionResult = ConstraintSuggestionRunner(spark) \
    .onData(df) \
    .addConstraintRule(DEFAULT()) \
    .run()
```

--------------------------------

### haveAnyCompleteness

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint that asserts on any completion in the combined set of columns.

```APIDOC
## haveAnyCompleteness

### Description
Creates a constraint that asserts on any completion in the combined set of columns.

### Method
Not applicable (function call)

### Endpoint
Not applicable (function call)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
check = haveAnyCompleteness(_columns_=['col1', 'col2'], _assertion_=lambda c: c > 0.95)
```

### Response
#### Success Response (200)
None (returns a Check object)

#### Response Example
None
```

--------------------------------

### PyDeequ Scala Utilities

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Utility functions for interacting with Scala objects and classes through Py4J, including type conversions and function wrappers.

```APIDOC
## Module pydeequ.scala_utils

A collection of utility functions and classes for manipulating with scala objects and classes through py4j.

## Class _pydeequ.scala_utils.PythonCallback

### Description
Base class for callback functions bridging Python and Scala.

### Parameters
* **gateway** - Py4J Gateway client.

### Bases
- `object`

## Class _pydeequ.scala_utils.ScalaFunction1

### Description
Implements scala.Function1 interface so we can pass lambda functions to Scala.

### Parameters
* **gateway** - Py4J Gateway client.
* **lambda_function** - The Python lambda function to wrap.

### Bases
- `PythonCallback`

#### Method `apply(_arg_)`
Implements the apply function.

## Class _pydeequ.scala_utils.ScalaFunction2

### Description
Implements scala.Function2 interface.

### Parameters
* **gateway** - Py4J Gateway client.
* **lambda_function** - The Python lambda function to wrap.

### Bases
- `PythonCallback`

#### Method `apply(_t1_, _t2_)`
Implements the apply function.

## Functions

#### `pydeequ.scala_utils.get_or_else_none(_scala_option_)`
Converts a Scala Option to a Python None or the contained value.

#### `pydeequ.scala_utils.java_list_to_python_list(_java_list: str_, _datatype_)`
Converts a Java List to a Python list.

#### `pydeequ.scala_utils.scala_get_default_argument(_java_object_: JavaObject, _argument_idx: int_) → JavaObject`
Retrieves the default argument value from a Scala object.

#### `pydeequ.scala_utils.scala_map_to_dict(_jvm_, _scala_map_)`
Converts a Scala Map to a Python dictionary.

#### `pydeequ.scala_utils.scala_map_to_java_map(_jvm_, _scala_map_)`
Converts a Python dictionary to a Scala Map.

#### `pydeequ.scala_utils.to_scala_map(_spark_session_, _d_) → Scala map`
Convert a dict into a JVM Map.
* **Args**:
  * `spark_session`: Spark session.
  * `d`: Python dictionary.
* **Returns**:
  Scala map.

#### `pydeequ.scala_utils.to_scala_seq(_jvm_, _iterable_) → Scala sequence`
Helper method to take an iterable and turn it into a Scala sequence.
* **Args**:
  * `jvm`: Spark session’s JVM.
  * `iterable`: Your iterable.
* **Returns**:
  Scala sequence.
```

--------------------------------

### isComplete

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint that asserts on a column completion.

```APIDOC
## isComplete

### Description
Creates a constraint that asserts on a column completion.

### Method
Not applicable (function call)

### Endpoint
Not applicable (function call)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
check = isComplete(_column_='my_column')
```

### Response
#### Success Response (200)
None (returns a Check object)

#### Response Example
None
```

--------------------------------

### Enforce Data Quality Checks

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Define and run data quality checks using the VerificationSuite and Check objects. This snippet demonstrates checks for size, minimum values, completeness, uniqueness, containment, and non-negativity.

```python
from pydeequ.checks import *
from pydeequ.verification import *

# Define data quality checks
check = Check(spark, CheckLevel.Warning, "Review Check")

checkResult = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(
        check.hasSize(lambda x: x >= 3) \
        .hasMin("b", lambda x: x == 1) \
        .isComplete("c") \
        .isUnique("a") \
        .isContainedIn("a", ["foo", "bar", "baz"]) \
        .isNonNegative("b") \
        .hasCompleteness("b", lambda x: x == 1.0)
    ) \
    .run()

# Convert results to DataFrame
checkResult_df = VerificationResult.checkResultsAsDataFrame(spark, checkResult)
checkResult_df.show(truncate=False)
```

--------------------------------

### Scala Utilities in pydeequ.scala_utils

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Provides utility functions for interacting with Scala objects and functions via py4j. This includes converting between Python and Scala data structures, handling Scala options, and implementing Scala's Function interfaces (Function1, Function2) using Python lambda functions.

```python
pydeequ.scala_utils.get_or_else_none(_scala_option_)

pydeequ.scala_utils.java_list_to_python_list(_java_list : str_, _datatype_)

pydeequ.scala_utils.scala_get_default_argument(_java_object_ , _argument_idx : int_) → JavaObject

pydeequ.scala_utils.scala_map_to_dict(_jvm_ , _scala_map_)

pydeequ.scala_utils.scala_map_to_java_map(_jvm_ , _scala_map_)

pydeequ.scala_utils.to_scala_map(_spark_session_ , _d_)
    Convert a dict into a JVM Map. Args:
> spark_session: Spark session d: Python dictionary 

Returns:
    
Scala map 

pydeequ.scala_utils.to_scala_seq(_jvm_ , _iterable_)
    Helper method to take an iterable and turn it into a Scala sequence Args:
> jvm: Spark session’s JVM iterable: your iterable 

Returns:
    
Scala sequence
```

--------------------------------

### Perform Comprehensive Data Profiling

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Utilize the ColumnProfilerRunner to automatically profile all columns in a Spark DataFrame. This provides insights into completeness, distinct values, and other characteristics for each column.

```python
from pydeequ.profiles import *

# Profile all columns automatically
result = ColumnProfilerRunner(spark) \
    .onData(df) \
    .run()

# Access individual column profiles
for col, profile in result.profiles.items():
    print(f"Column: {col}")
    print(f"Completeness: {profile.completeness}")
    print(f"Approximate distinct count: {profile.approximateNumDistinctValues}")
    print(profile)
```

--------------------------------

### Compute Metrics with PyDeequ Analyzers

Source: https://pydeequ.readthedocs.io/en/latest/README

Uses PyDeequ's `AnalysisRunner` to compute various data quality metrics on a Spark DataFrame. It demonstrates adding analyzers like `Size` and `Completeness` and converting the results to a DataFrame.

```python
from pydeequ.analyzers import *

analysisResult = AnalysisRunner(spark) \
                    .onData(df) \
                    .addAnalyzer(Size()) \
                    .addAnalyzer(Completeness("b")) \
                    .run()

analysisResult_df = AnalyzerContext.successMetricsAsDataFrame(spark, analysisResult)
anlysisResult_df.show()

```

--------------------------------

### hasCompleteness

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint that asserts column completion using historical completeness values.

```APIDOC
## hasCompleteness

### Description
Creates a constraint that asserts column completion. Uses the given history selection strategy to retrieve historical completeness values on this column from the history provider.

### Method
[Assumed POST or PUT based on constraint creation]

### Endpoint
[Assumed /check/hasCompleteness]

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **column** (str) - Required - Column in Data Frame to run the assertion on.
- **assertion** (lambda) - Required - A function that accepts an int or float parameter.
- **hint** (str) - Optional - A hint that states why a constraint could have failed.

### Request Example
```json
{
  "column": "user_id",
  "assertion": "lambda x: x > 0.95",
  "hint": "Completeness below threshold"
}
```

### Response
#### Success Response (200)
- **check_object** (object) - A Check object that implements the assertion.

#### Response Example
```json
{
  "check_object": { "type": "hasCompleteness", "column": "user_id", "assertion": "lambda x: x > 0.95" }
}
```
```

--------------------------------

### haveCompleteness

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint that asserts on completed rows in a combined set of columns.

```APIDOC
## haveCompleteness

### Description
Creates a constraint that asserts on completed rows in a combined set of columns.

### Method
Not applicable (function call)

### Endpoint
Not applicable (function call)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
check = haveCompleteness(_columns_=['col1', 'col2'], _assertion_=lambda c: c == 1.0)
```

### Response
#### Success Response (200)
None (returns a Check object)

#### Response Example
None
```

--------------------------------

### PyDeequ Analyzers API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

This section details the various analyzers available in PyDeequ for data quality assessment.

```APIDOC
## PyDeequ Analyzers API

### Description
Provides a comprehensive list of analyzers for data quality metrics.

### Analyzers
- `AnalysisRunBuilder`
- `AnalysisRunner`
- `AnalyzerContext`
- `ApproxCountDistinct`
- `ApproxQuantile`
- `ApproxQuantiles`
- `Completeness`
- `Compliance`
- `Correlation`
- `CountDistinct`
- `DataType`
- `DataTypeInstances`
- `Distinctness`
- `Entropy`
- `Histogram`
- `KLLParameters`
- `KLLSketch`
- `MaxLength`
- `Maximum`
- `Mean`
- `MinLength`
- `Minimum`
- `MutualInformation`
- `PatternMatch`
- `Size`
- `StandardDeviation`
- `Sum`
- `UniqueValueRatio`
- `Uniqueness`
```

--------------------------------

### Generate Column Profiles with PyDeequ

Source: https://pydeequ.readthedocs.io/en/latest/README

Utilizes `ColumnProfilerRunner` to generate detailed profiles for each column in a Spark DataFrame. The output includes statistics and information about each column's data.

```python
from pydeequ.profiles import *

result = ColumnProfilerRunner(spark)
    .onData(df)
    .run()

for col, profile in result.profiles.items():
    print(profile)

```

--------------------------------

### InMemoryMetricsRepository Class in pydeequ.repository

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Represents a high-level interface for an in-memory metrics repository. This class is suitable for temporary storage and retrieval of analysis results within the current Spark session. It requires a SparkSession object for its initialization.

```python
class _pydeequ.repository.InMemoryMetricsRepository(_spark_session : SparkSession_)
    Bases: MetricsRepository
    High level InMemoryMetricsRepository Interface 
```

--------------------------------

### Utilize Specialized Analyzers in PyDeequ

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Demonstrates the application of various specialized analyzers provided by PyDeequ for in-depth data profiling. This includes calculating approximate quantiles, distinctness fraction, generating histograms, measuring mutual information, and pattern matching.

```python
from pydeequ.analyzers import *

# ApproxQuantile - Calculate percentiles
analysis = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(ApproxQuantile("b", 0.5))  # Median \
    .addAnalyzer(ApproxQuantile("b", 0.95))  # 95th percentile \
    .run()

# Distinctness - Fraction of distinct values
analysis = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(Distinctness(["a", "b"]))  # Multi-column distinctness \
    .run()

# Histogram - Value distribution
analysis = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(Histogram("a", maxDetailBins=10)) \
    .run()

# MutualInformation - Information dependency between columns
analysis = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(MutualInformation(["a", "b"])) \
    .run()

# PatternMatch - Regex compliance
analysis = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(PatternMatch("a", r"^[a-z]{3}$")) \
    .run()
```

--------------------------------

### Compute Data Quality Metrics using Analyzers

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Use the AnalysisRunner to compute various data quality metrics on a Spark DataFrame. This includes metrics like size, completeness, mean, distinct count, and correlation.

```python
from pydeequ.analyzers import *

# Run multiple analyzers to compute various metrics
analysisResult = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(Size()) \
    .addAnalyzer(Completeness("b")) \
    .addAnalyzer(Mean("b")) \
    .addAnalyzer(ApproxCountDistinct("a")) \
    .addAnalyzer(Correlation("b", "c")) \
    .run()

# Convert results to DataFrame
analysisResult_df = AnalyzerContext.successMetricsAsDataFrame(spark, analysisResult)
analysisResult_df.show()
```

--------------------------------

### Implement Advanced Data Validation Constraints

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Utilize advanced check constraints for more sophisticated data validation, including pattern matching, email and credit card number detection, entropy, correlation, and date comparisons.

```python
from pydeequ.checks import *
from pydeequ.verification import *

# Pattern matching and compliance checks
check = Check(spark, CheckLevel.Error, "Data Validation")

verification = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(
        check
        .containsEmail("email_col", lambda x: x >= 0.9)
        .containsCreditCardNumber("cc_col", lambda x: x == 0.0)
        .hasPattern("phone_col", r"^\d{3}-\d{3}-\d{4}$")
        .hasEntropy("category_col", lambda x: x > 2.0)
        .hasCorrelation("col1", "col2", lambda x: x > 0.5)
        .isGreaterThan("start_date", "end_date")
    ) \
    .run()
```

--------------------------------

### PyDeequ Anomaly Detection API

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Details related to anomaly detection capabilities within PyDeequ.

```APIDOC
## PyDeequ Anomaly Detection API

### Description
Information on anomaly detection modules and functionalities.
```

--------------------------------

### Perform Conditional Analysis with WHERE Clauses

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Shows how to apply filters directly within the PyDeequ analysis pipeline using the 'where' clause. This allows for targeted analysis on subsets of data without pre-filtering the DataFrame.

```python
from pydeequ.analyzers import *

# Apply filters before analysis
analysisResult = AnalysisRunner(spark) \
    .onData(df) \
    .addAnalyzer(Completeness("c", where="b > 1")) \
    .addAnalyzer(Mean("b", where="c IS NOT NULL")) \
    .addAnalyzer(Size(where="a IN ('foo', 'bar')")) \
    .run()

analysisResult_df = AnalyzerContext.successMetricsAsDataFrame(spark, analysisResult)
```

--------------------------------

### Validate Data Types using PyDeequ Checks

Source: https://context7.com/context7/pydeequ_readthedocs_io_en/llms.txt

Demonstrates how to enforce data type constraints on columns using PyDeequ's `hasDataType` check. It covers validation for String, Integral, and Fractional data types.

```python
from pydeequ.checks import *
from pydeequ.verification import *

check = Check(spark, CheckLevel.Error, "Type Validation")

verification = VerificationSuite(spark) \
    .onData(df) \
    .addCheck(
        check
        .hasDataType("a", ConstrainableDataTypes.String)
        .hasDataType("b", ConstrainableDataTypes.Integral)
        .hasDataType("price", ConstrainableDataTypes.Fractional)
    ) \
    .run()
```

--------------------------------

### hasUniqueValueRatio

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint on the unique value ratio in a single or combined set of key columns.

```APIDOC
## hasUniqueValueRatio

### Description
Creates a constraint on the unique value ratio in a single or combined set of key columns.

### Method
Not applicable (function call)

### Endpoint
Not applicable (function call)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
check = hasUniqueValueRatio(_columns_=['col1', 'col2'], _assertion_=lambda r: r > 0.9)
```

### Response
#### Success Response (200)
None (returns a Check object)

#### Response Example
None
```

--------------------------------

### ResultKey Class in pydeequ.repository

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Encapsulates information that uniquely identifies an AnalysisResult within the repository. It includes the SparkSession, dataset date, and a map of tags for additional annotations. This class is crucial for referencing specific results.

```python
class _pydeequ.repository.ResultKey(_spark_session : SparkSession_, _dataSetDate : Optional[int] = None_, _tags : Optional[dict] = None_)
    Bases: object
    Information that uniquely identifies a AnalysisResult 

    Parameters:
        
      * **spark_session** – SparkSession
      * **dataSetDate** – Date of the result key
      * **tags** – A map with additional annotations

    _static _current_milli_time()
        Get current time in milliseconds # TODO: Consider putting this into scala_utils? Or general utils?
```

--------------------------------

### hasMax

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint that asserts on the maximum of the column.

```APIDOC
## hasMax

### Description
Creates a constraint that asserts on the maximum of the column. The column contains either a long, int or float datatype.

### Method
[Assumed POST or PUT based on constraint creation]

### Endpoint
[Assumed /check/hasMax]

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **column** (str) - Required - Column in Data Frame to run the assertion on. The column is expected to be an int, long or float type.
- **assertion** (lambda) - Required - A function which accepts an int or float parameter that discerns the maximum value of the column.
- **hint** (str) - Optional - A hint that states why a constraint could have failed.

### Request Example
```json
{
  "column": "price",
  "assertion": "lambda x: x < 1000.0",
  "hint": "Maximum price exceeded"
}
```

### Response
#### Success Response (200)
- **check_object** (object) - A Check object that asserts maximum of the column.

#### Response Example
```json
{
  "check_object": { "type": "hasMax", "column": "price", "assertion": "lambda x: x < 1000.0" }
}
```
```

--------------------------------

### hasSum

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint that asserts on the sum of the column.

```APIDOC
## hasSum

### Description
Creates a constraint that asserts on the sum of the column.

### Method
Not applicable (function call)

### Endpoint
Not applicable (function call)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
check = hasSum(_column_='my_column', _assertion_=lambda s: s > 0)
```

### Response
#### Success Response (200)
None (returns a Check object)

#### Response Example
None
```

--------------------------------

### ScalaFunction2 Class in pydeequ.scala_utils

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Implements the scala.Function2 interface, enabling Python lambda functions that accept two arguments to be passed to Scala methods. It extends PythonCallback and provides an apply method to handle the two arguments as expected by Scala.

```python
class _pydeequ.scala_utils.ScalaFunction2(_gateway_ , _lambda_function_)
    Bases: PythonCallback
Implements scala.Function2 interface https://www.scala-lang.org/api/current/scala/Function2.html 

class _Java
    Bases: object
scala.Function2: a function that takes two arguments 

    implements _ = ['scala.Function2']_
     

    apply(_t1_ , _t2_)
        Implements the apply function 
```

--------------------------------

### MetricsRepository Base Class in pydeequ.repository

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

The base class for all metrics repository implementations in pydeequ. It defines common methods for filtering and retrieving analysis results based on various criteria such as time, analyzers, and tags. This class serves as an abstract foundation for specific repository types.

```python
class _pydeequ.repository.MetricsRepository
    Bases: object
    Base class for Metrics Repository 

    after(_dateTime : int_)
        Only look at AnalysisResults with a result key with a greater value :param dateTime: The minimum dateTime of AnalysisResults to look at 

    before(_dateTime : int_)
        Only look at AnalysisResults with a result key with a smaller value :param dateTime: The maximum dateTime of AnalysisResults to look at 

    forAnalyzers(_analyzers : list_)
        Choose all metrics that you want to load :param analyzers: List of analyers who’s resulting metrics you want to load 

    getSuccessMetricsAsDataFrame(_withTags : Optional[list] = None_, _pandas : bool = False_)
        Get the AnalysisResult as DataFrame :param withTags: List of tags to filter previous Metrics Repository runs with 

    getSuccessMetricsAsJson(_withTags : Optional[list] = None_)
        Get the AnalysisResult as JSON :param withTags: List of tags to filter previous Metrics Repository runs with 

    _classmethod _helper_metrics_file(_spark_session : SparkSession_, _filename : str = 'metrics.json'_)
        Helper method to create the metrics file for storage 

    load()
        Get a builder class to construct a loading query to get AnalysisResults 

    withTagValues(_tagValues : dict_)
        Filter out results that don’t have specific values for specific tags :param tagValues: Dict with tag names and the corresponding values to filter for 
```

--------------------------------

### Compute Approximate Quantiles in PyDeequ

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Describes the ApproxQuantiles analyzer, which computes multiple approximate quantiles for a given column. It allows specifying a list of quantiles and a relative error.

```python
from pydeequ.analyzers import ApproxQuantiles

# Compute approximate 25th, 50th, and 75th percentiles
quantiles_analyzer = ApproxQuantiles(column="score", quantiles=[0.25, 0.5, 0.75])

# With a custom relative error
# precise_quantiles = ApproxQuantiles(column="score", quantiles=[0.1, 0.9], relativeError=0.001)
```

--------------------------------

### hasEntropy

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint that asserts on a column's entropy.

```APIDOC
## hasEntropy

### Description
Creates a constraint that asserts on a column entropy. Entropy is a measure of the level of information contained in a message.

### Method
[Assumed POST or PUT based on constraint creation]

### Endpoint
[Assumed /check/hasEntropy]

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **column** (str) - Required - Column in Data Frame to run the assertion on.
- **assertion** (lambda) - Required - A function that accepts an int or float parameter.
- **hint** (str) - Optional - A hint that states why a constraint could have failed.

### Request Example
```json
{
  "column": "session_id",
  "assertion": "lambda x: x > 1.5",
  "hint": "Low entropy detected"
}
```

### Response
#### Success Response (200)
- **check_object** (object) - A Check object that asserts the entropy in the column.

#### Response Example
```json
{
  "check_object": { "type": "hasEntropy", "column": "session_id", "assertion": "lambda x: x > 1.5" }
}
```
```

--------------------------------

### Assert Column Sketch Size

Source: https://pydeequ.readthedocs.io/en/latest/pydeequ

Creates a constraint to assert the size of a column's sketch. This function requires the column name, an assertion function that operates on the sketch distribution, and optional KLL sketch parameters and a hint. It returns a Check object for sketch size assertion.

```python
def kllSketchSatisfies(_column_ , _assertion_ , _kllParameters =None_, _hint =None_):
    """
    Creates a constraint that asserts on column’s sketch size. 

    Parameters:
    * **column** (_str_) – Column in Data Frame to run the assertion on.
    * **assertion** (_Lambda_ _(__BucketDistribution_ _)_) – A function that accepts an int or float parameter as a distribution input parameter.
    * **kllParameters** (_KLLParameters_) – Parameters of KLL sketch
    * **hint** (_str_) – A hint that states why a constraint could have failed.


    Returns:
    kllSketchSatisfies self: A Check object that asserts the column’s sketch size in the column.
    """
    pass
```
Tokens:
Raw