Skip to content

Data validator

DataValidator dataclass

A class to validate data based on a set of validation rules.

Attributes:

Name Type Description
rules list[ValidationRule]

A list of validation rules.
Example 
>>> ValidationRules = [
...     ValidationRule("first_name_check", F_.is_printable_only("first_name")),
...     ValidationRule("last_name_check", F_.is_printable_only("last_name")),
...     ValidationRule("address_check", F_.is_printable_only("address")),
...     ValidationRule("region_check", F_.is_printable_only("region")),
...     ValidationRule("code_check", [F_.is_two_character_only("code")]),
...     ValidationRule("postcode_check", F_.is_printable_only("postcode")),
... ]

>>> validator = DataValidator(ValidationRules)

>>> conditions = {
...     "first_name_check": F_.is_printable_only("first_name"),
...     "last_name_check": F_.is_printable_only("last_name"),
...     "address_check": F_.is_printable_only("address"),
...     "region_check": F_.is_printable_only("region"),
...     "code_check": [F_.is_two_character_only("code")],
...     "postcode_check": F_.is_printable_only("postcode"),
... }

>>> validator = DataValidator.from_dict(conditions)
Source code in pysparky/data_validator.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@dataclass
class DataValidator:
    """
    A class to validate data based on a set of validation rules.

    Attributes:
        rules (list[ValidationRule]): A list of validation rules.

    Example:
        ```python
        >>> ValidationRules = [
        ...     ValidationRule("first_name_check", F_.is_printable_only("first_name")),
        ...     ValidationRule("last_name_check", F_.is_printable_only("last_name")),
        ...     ValidationRule("address_check", F_.is_printable_only("address")),
        ...     ValidationRule("region_check", F_.is_printable_only("region")),
        ...     ValidationRule("code_check", [F_.is_two_character_only("code")]),
        ...     ValidationRule("postcode_check", F_.is_printable_only("postcode")),
        ... ]

        >>> validator = DataValidator(ValidationRules)

        >>> conditions = {
        ...     "first_name_check": F_.is_printable_only("first_name"),
        ...     "last_name_check": F_.is_printable_only("last_name"),
        ...     "address_check": F_.is_printable_only("address"),
        ...     "region_check": F_.is_printable_only("region"),
        ...     "code_check": [F_.is_two_character_only("code")],
        ...     "postcode_check": F_.is_printable_only("postcode"),
        ... }

        >>> validator = DataValidator.from_dict(conditions)
        ```
    """

    rules: list[ValidationRule]

    @classmethod
    def from_dict(cls, data: dict[str, Union[list[Column], Column]]) -> "DataValidator":
        """
        Creates a DataValidator instance from a dictionary.

        Args:
            data (dict[str, list[Column] | Column]): A dictionary where keys are rule names and values are lists of conditions or a single condition.

        Returns:
            DataValidator: An instance of DataValidator.

        Example:
            ```python
            >>> conditions = {
            ...     "first_name_check": F_.is_printable_only("first_name"),
            ...     "last_name_check": F_.is_printable_only("last_name"),
            ...     "address_check": F_.is_printable_only("address"),
            ...     "region_check": F_.is_printable_only("region"),
            ...     "code_check": [F_.is_two_character_only("code")],
            ...     "postcode_check": F_.is_printable_only("postcode"),
            ... }
            >>> validator = DataValidator.from_dict(conditions)
            ```

        """
        rules = [ValidationRule(name, conditions) for name, conditions in data.items()]
        return cls(rules=rules)

    @property
    def query_map(self) -> dict[str, Column]:
        """
        Gets a dictionary mapping rule names to their combined conditions.
        The key is the column name and the value is the Column Object.

        Returns:
            dict[str, Column]: A dictionary where keys are rule names
                and values are combined conditions.

        Example:
            ```python
            >>> sdf.withColumns(validator.query_map)
            ```

        """
        return {rule.name: rule.combined_condition for rule in self.rules}

    def apply_conditions(self, sdf: DataFrame) -> DataFrame:
        """
        Applies the combined conditions to the Spark DataFrame.

        Args:
            sdf (DataFrame): The Spark DataFrame to which the conditions will be applied.

        Returns:
            DataFrame: The Spark DataFrame with the conditions applied.

        Example:
            ```python
            >>> validator.apply_conditions(data_sdf)
            ```

        """
        return sdf.withColumns(self.query_map)

    def filter_invalid(self, sdf: DataFrame) -> DataFrame:
        """
        Filters out invalid rows from the Spark DataFrame based on the rules.

        Args:
            sdf (DataFrame): The Spark DataFrame to be filtered.

        Returns:
            DataFrame: The Spark DataFrame with invalid rows filtered out.

        Example:
            ```python
            >>> validator.filter_invalid(data_sdf)
            ```

        """
        return filters(
            self.apply_conditions(sdf),
            [
                (
                    F.col(column_name) == False  # noqa: E712
                )  # pylint: disable=singleton-comparison
                for column_name in self.query_map.keys()
            ],
            operator_="or",
        )

    def filter_valid(self, sdf: DataFrame) -> DataFrame:
        """
        Filters out valid rows from the Spark DataFrame based on the rules.

        Args:
            sdf (DataFrame): The Spark DataFrame to be filtered.

        Returns:
            DataFrame: The Spark DataFrame with valid rows filtered out.

        Example:
            ```python
            >>> validator.filter_valid(data_sdf).select(data_sdf.columns).show()
            ```

        """
        return filters(
            self.apply_conditions(sdf),
            [
                (
                    F.col(column_name) == True  # noqa: E712
                )  # noqa: E712 # pylint: disable=singleton-comparison
                for column_name in self.query_map.keys()
            ],
            operator_="and",
        )

query_map property

Gets a dictionary mapping rule names to their combined conditions. The key is the column name and the value is the Column Object.

Returns:

Type Description
dict[str, Column]

dict[str, Column]: A dictionary where keys are rule names and values are combined conditions.
Example 
>>> sdf.withColumns(validator.query_map)

apply_conditions(sdf)

Applies the combined conditions to the Spark DataFrame.

Parameters:

Name Type Description Default
sdf DataFrame

The Spark DataFrame to which the conditions will be applied.

 required

Returns:

Name Type Description
DataFrame DataFrame

The Spark DataFrame with the conditions applied.
Example 
>>> validator.apply_conditions(data_sdf)
Source code in pysparky/data_validator.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def apply_conditions(self, sdf: DataFrame) -> DataFrame:
    """
    Applies the combined conditions to the Spark DataFrame.

    Args:
        sdf (DataFrame): The Spark DataFrame to which the conditions will be applied.

    Returns:
        DataFrame: The Spark DataFrame with the conditions applied.

    Example:
        ```python
        >>> validator.apply_conditions(data_sdf)
        ```

    """
    return sdf.withColumns(self.query_map)

filter_invalid(sdf)

Filters out invalid rows from the Spark DataFrame based on the rules.

Parameters:

Name Type Description Default
sdf DataFrame

The Spark DataFrame to be filtered.

 required

Returns:

Name Type Description
DataFrame DataFrame

The Spark DataFrame with invalid rows filtered out.
Example 
>>> validator.filter_invalid(data_sdf)
Source code in pysparky/data_validator.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def filter_invalid(self, sdf: DataFrame) -> DataFrame:
    """
    Filters out invalid rows from the Spark DataFrame based on the rules.

    Args:
        sdf (DataFrame): The Spark DataFrame to be filtered.

    Returns:
        DataFrame: The Spark DataFrame with invalid rows filtered out.

    Example:
        ```python
        >>> validator.filter_invalid(data_sdf)
        ```

    """
    return filters(
        self.apply_conditions(sdf),
        [
            (
                F.col(column_name) == False  # noqa: E712
            )  # pylint: disable=singleton-comparison
            for column_name in self.query_map.keys()
        ],
        operator_="or",
    )

filter_valid(sdf)

Filters out valid rows from the Spark DataFrame based on the rules.

Parameters:

Name Type Description Default
sdf DataFrame

The Spark DataFrame to be filtered.

 required

Returns:

Name Type Description
DataFrame DataFrame

The Spark DataFrame with valid rows filtered out.
Example 
>>> validator.filter_valid(data_sdf).select(data_sdf.columns).show()
Source code in pysparky/data_validator.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def filter_valid(self, sdf: DataFrame) -> DataFrame:
    """
    Filters out valid rows from the Spark DataFrame based on the rules.

    Args:
        sdf (DataFrame): The Spark DataFrame to be filtered.

    Returns:
        DataFrame: The Spark DataFrame with valid rows filtered out.

    Example:
        ```python
        >>> validator.filter_valid(data_sdf).select(data_sdf.columns).show()
        ```

    """
    return filters(
        self.apply_conditions(sdf),
        [
            (
                F.col(column_name) == True  # noqa: E712
            )  # noqa: E712 # pylint: disable=singleton-comparison
            for column_name in self.query_map.keys()
        ],
        operator_="and",
    )

from_dict(data) classmethod

Creates a DataValidator instance from a dictionary.

Parameters:

Name Type Description Default
data dict[str, list[Column] | Column]

A dictionary where keys are rule names and values are lists of conditions or a single condition.

 required

Returns:

Name Type Description
DataValidator DataValidator

An instance of DataValidator.
Example 
>>> conditions = {
...     "first_name_check": F_.is_printable_only("first_name"),
...     "last_name_check": F_.is_printable_only("last_name"),
...     "address_check": F_.is_printable_only("address"),
...     "region_check": F_.is_printable_only("region"),
...     "code_check": [F_.is_two_character_only("code")],
...     "postcode_check": F_.is_printable_only("postcode"),
... }
>>> validator = DataValidator.from_dict(conditions)
Source code in pysparky/data_validator.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@classmethod
def from_dict(cls, data: dict[str, Union[list[Column], Column]]) -> "DataValidator":
    """
    Creates a DataValidator instance from a dictionary.

    Args:
        data (dict[str, list[Column] | Column]): A dictionary where keys are rule names and values are lists of conditions or a single condition.

    Returns:
        DataValidator: An instance of DataValidator.

    Example:
        ```python
        >>> conditions = {
        ...     "first_name_check": F_.is_printable_only("first_name"),
        ...     "last_name_check": F_.is_printable_only("last_name"),
        ...     "address_check": F_.is_printable_only("address"),
        ...     "region_check": F_.is_printable_only("region"),
        ...     "code_check": [F_.is_two_character_only("code")],
        ...     "postcode_check": F_.is_printable_only("postcode"),
        ... }
        >>> validator = DataValidator.from_dict(conditions)
        ```

    """
    rules = [ValidationRule(name, conditions) for name, conditions in data.items()]
    return cls(rules=rules)

ValidationRule dataclass

A class to represent a validation rule.

Attributes:

Name Type Description
name str

The name of the validation rule.
conditions list[Column]

A list of conditions (Spark Columns) that make up the rule.
combined_condition Column

The combined condition of all the conditions using logical AND. It will generate from conditions
Example

ValidationRule("first_name_check", F_.printable_only("first_name")),

Source code in pysparky/data_validator.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@dataclass
class ValidationRule:
    """
    A class to represent a validation rule.

    Attributes:
        name (str): The name of the validation rule.
        conditions (list[Column]): A list of conditions (Spark Columns) that make up the rule.
        combined_condition (Column): The combined condition of all the conditions using logical AND.
            It will generate from conditions

    Example:
        ValidationRule("first_name_check", F_.printable_only("first_name")),
    """

    name: str
    conditions: list[Column] | Column
    combined_condition: Column = field(init=False)

    def __post_init__(self):
        self.conditions = enabler.ensure_list(self.conditions)
        self.combined_condition = F_.condition_and(*self.conditions)