20220912.22

Build # 20220912.22

Build Type

push

circleci

Committed by GitHub

Commit Message

<a href="https://github.com/great-expectations/great_expectations/commit/<a class=hub.com/great-expectations/great_expectations/commit/f89ea99db8f42554ce6112d675c06b7a49a1ae80">f89ea99db<a href="https://github.com/great-expectations/great_expectations/commit/f89ea99db8f42554ce6112d675c06b7a49a1ae80">">Merge </a><a class="double-link" href="https://github.com/great-expectations/great_expectations/commit/<a class="double-link" href="https://github.com/great-expectations/great_expectations/commit/2bf486d3454570be72141d69a21e19641a408be5">2bf486d34</a>">2bf486d34</a><a href="https://github.com/great-expectations/great_expectations/commit/f89ea99db8f42554ce6112d675c06b7a49a1ae80"> into b37cacc2f">b37cacc2f</a>

Run Details

19008 of 49134 relevant lines covered (38.69%)

1.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.16

/great_expectations/expectations/metrics/column_aggregate_metrics/column_distinct_values.py

from typing import Any, Dict, List, Optional, Set

import pandas as pd

from great_expectations.core import ExpectationConfiguration
from great_expectations.core.metric_domain_types import MetricDomainTypes
from great_expectations.execution_engine import (
    ExecutionEngine,
    PandasExecutionEngine,
    SparkDFExecutionEngine,
    SqlAlchemyExecutionEngine,
)
from great_expectations.expectations.metrics.column_aggregate_metric_provider import (
    ColumnAggregateMetricProvider,
    column_aggregate_partial,
    column_aggregate_value,
)
from great_expectations.expectations.metrics.import_manager import (
    F,
    pyspark_sql_Column,
    pyspark_sql_DataFrame,
    pyspark_sql_Row,
    sa,
    sa_func_count,
    sa_sql_expression_ColumnClause,
    sa_sql_expression_Selectable,
    sqlalchemy_engine_Engine,
)
from great_expectations.expectations.metrics.metric_provider import metric_value
from great_expectations.validator.metric_configuration import MetricConfiguration


class ColumnDistinctValues(ColumnAggregateMetricProvider):
    metric_name = "column.distinct_values"

    @column_aggregate_value(engine=PandasExecutionEngine)
    def _pandas(cls, column: pd.Series, **kwargs) -> Set[Any]:
        return set(column.unique())

    @metric_value(engine=SqlAlchemyExecutionEngine)
    def _sqlalchemy(
        cls,
        execution_engine: SqlAlchemyExecutionEngine,
        metric_domain_kwargs: Dict[str, str],
        **kwargs,
    ) -> Set[Any]:
        """
        Past implementations of column.distinct_values depended on column.value_counts.
        This was causing performance issues due to the complex query used in column.value_counts and subsequent
        in-memory operations.
        """
        selectable: sa_sql_expression_Selectable
        accessor_domain_kwargs: Dict[str, str]
        (selectable, _, accessor_domain_kwargs,) = execution_engine.get_compute_domain(
            metric_domain_kwargs, MetricDomainTypes.COLUMN
        )
        column_name: str = accessor_domain_kwargs["column"]
        column: sa_sql_expression_ColumnClause = sa.column(column_name)
        sqlalchemy_engine = execution_engine.engine

        distinct_values: List[sqlalchemy_engine_Engine] = sqlalchemy_engine.execute(
            sa.select([column])
            .where(column.is_not(None))
            .distinct()
            .select_from(selectable)
        ).fetchall()
        # Vectorized operation is not faster here due to overhead of converting to and from numpy array
        return {row[0] for row in distinct_values}

    @metric_value(engine=SparkDFExecutionEngine)
    def _spark(
        cls,
        execution_engine: SparkDFExecutionEngine,
        metric_domain_kwargs: Dict[str, str],
        **kwargs,
    ) -> Set[Any]:
        """
        Past implementations of column.distinct_values depended on column.value_counts.
        This was causing performance issues due to the complex query used in column.value_counts and subsequent
        in-memory operations.
        """
        df: pyspark_sql_DataFrame
        accessor_domain_kwargs: Dict[str, str]
        (df, _, accessor_domain_kwargs,) = execution_engine.get_compute_domain(
            metric_domain_kwargs, MetricDomainTypes.COLUMN
        )
        column_name: str = accessor_domain_kwargs["column"]
        distinct_values: List[pyspark_sql_Row] = (
            df.select(F.col(column_name))
            .distinct()
            .where(F.col(column_name).isNotNull())
            .rdd.flatMap(lambda x: x)
            .collect()
        )
        return set(distinct_values)


class ColumnDistinctValuesCount(ColumnAggregateMetricProvider):
    metric_name = "column.distinct_values.count"

    @column_aggregate_value(engine=PandasExecutionEngine)
    def _pandas(cls, column: pd.Series, **kwargs) -> int:
        return column.nunique()

    @column_aggregate_partial(engine=SqlAlchemyExecutionEngine)
    def _sqlalchemy(
        cls,
        column: sa_sql_expression_ColumnClause,
        **kwargs,
    ) -> sa_func_count:
        """
        Past implementations of column.distinct_values.count depended on column.value_counts and column.distinct_values.
        This was causing performance issues due to the complex query used in column.value_counts and subsequent
        in-memory operations.
        """
        return sa.func.count(sa.distinct(column))

    @column_aggregate_partial(engine=SparkDFExecutionEngine)
    def _spark(
        cls,
        column: pyspark_sql_Column,
        **kwargs,
    ) -> pyspark_sql_Column:
        """
        Past implementations of column.distinct_values.count depended on column.value_counts and column.distinct_values.
        This was causing performance issues due to the complex query used in column.value_counts and subsequent
        in-memory operations.
        """
        return F.countDistinct(column)


class ColumnDistinctValuesCountUnderThreshold(ColumnAggregateMetricProvider):
    metric_name = "column.distinct_values.count.under_threshold"
    condition_keys = ("threshold",)

    @column_aggregate_value(engine=PandasExecutionEngine)
    def _pandas(cls, column: pd.Series, threshold: int, **kwargs) -> bool:
        return column.nunique() < threshold

    @metric_value(engine=SqlAlchemyExecutionEngine)
    def _sqlalchemy(
        cls,
        metric_value_kwargs: Dict[str, int],
        metrics: Dict[str, int],
        **kwargs,
    ) -> bool:
        return (
            metrics["column.distinct_values.count"] < metric_value_kwargs["threshold"]
        )

    @metric_value(engine=SparkDFExecutionEngine)
    def _spark(
        cls,
        metric_value_kwargs: Dict[str, int],
        metrics: Dict[str, int],
        **kwargs,
    ) -> bool:
        return (
            metrics["column.distinct_values.count"] < metric_value_kwargs["threshold"]
        )

    @classmethod
    def _get_evaluation_dependencies(
        cls,
        metric: MetricConfiguration,
        configuration: Optional[ExpectationConfiguration] = None,
        execution_engine: Optional[ExecutionEngine] = None,
        runtime_configuration: Optional[Dict] = None,
    ):
        """Returns a dictionary of given metric names and their corresponding configuration,
        specifying the metric types and their respective domains"""
        dependencies: dict = super()._get_evaluation_dependencies(
            metric=metric,
            configuration=configuration,
            execution_engine=execution_engine,
            runtime_configuration=runtime_configuration,
        )
        if metric.metric_name == "column.distinct_values.count.under_threshold":
            dependencies["column.distinct_values.count"] = MetricConfiguration(
                metric_name="column.distinct_values.count",
                metric_domain_kwargs=metric.metric_domain_kwargs,
                metric_value_kwargs=None,
            )
        return dependencies

1	from typing import Any, Dict, List, Optional, Set	3✔
2
3	import pandas as pd	3✔
4
5	from great_expectations.core import ExpectationConfiguration	3✔
6	from great_expectations.core.metric_domain_types import MetricDomainTypes	3✔
7	from great_expectations.execution_engine import (	3✔
8	ExecutionEngine,
9	PandasExecutionEngine,
10	SparkDFExecutionEngine,
11	SqlAlchemyExecutionEngine,
12	)
13	from great_expectations.expectations.metrics.column_aggregate_metric_provider import (	3✔
14	ColumnAggregateMetricProvider,
15	column_aggregate_partial,
16	column_aggregate_value,
17	)
18	from great_expectations.expectations.metrics.import_manager import (	3✔
19	F,
20	pyspark_sql_Column,
21	pyspark_sql_DataFrame,
22	pyspark_sql_Row,
23	sa,
24	sa_func_count,
25	sa_sql_expression_ColumnClause,
26	sa_sql_expression_Selectable,
27	sqlalchemy_engine_Engine,
28	)
29	from great_expectations.expectations.metrics.metric_provider import metric_value	3✔
30	from great_expectations.validator.metric_configuration import MetricConfiguration	3✔
31
32
33	class ColumnDistinctValues(ColumnAggregateMetricProvider):	3✔
34	metric_name = "column.distinct_values"	3✔
35
36	@column_aggregate_value(engine=PandasExecutionEngine)	3✔
37	def _pandas(cls, column: pd.Series, **kwargs) -> Set[Any]:	3✔
38	return set(column.unique())	×
39
40	@metric_value(engine=SqlAlchemyExecutionEngine)	3✔
41	def _sqlalchemy(	3✔
42	cls,
43	execution_engine: SqlAlchemyExecutionEngine,
44	metric_domain_kwargs: Dict[str, str],
45	**kwargs,
46	) -> Set[Any]:
47	"""
48	Past implementations of column.distinct_values depended on column.value_counts.
49	This was causing performance issues due to the complex query used in column.value_counts and subsequent
50	in-memory operations.
51	"""
52	selectable: sa_sql_expression_Selectable
53	accessor_domain_kwargs: Dict[str, str]
54	(selectable, _, accessor_domain_kwargs,) = execution_engine.get_compute_domain(	×
55	metric_domain_kwargs, MetricDomainTypes.COLUMN
56	)
57	column_name: str = accessor_domain_kwargs["column"]	×
58	column: sa_sql_expression_ColumnClause = sa.column(column_name)	×
59	sqlalchemy_engine = execution_engine.engine	×
60
61	distinct_values: List[sqlalchemy_engine_Engine] = sqlalchemy_engine.execute(	×
62	sa.select([column])
63	.where(column.is_not(None))
64	.distinct()
65	.select_from(selectable)
66	).fetchall()
67	# Vectorized operation is not faster here due to overhead of converting to and from numpy array
68	return {row[0] for row in distinct_values}	×
69
70	@metric_value(engine=SparkDFExecutionEngine)	3✔
71	def _spark(	3✔
72	cls,
73	execution_engine: SparkDFExecutionEngine,
74	metric_domain_kwargs: Dict[str, str],
75	**kwargs,
76	) -> Set[Any]:
77	"""
78	Past implementations of column.distinct_values depended on column.value_counts.
79	This was causing performance issues due to the complex query used in column.value_counts and subsequent
80	in-memory operations.
81	"""
82	df: pyspark_sql_DataFrame
83	accessor_domain_kwargs: Dict[str, str]
84	(df, _, accessor_domain_kwargs,) = execution_engine.get_compute_domain(	×
85	metric_domain_kwargs, MetricDomainTypes.COLUMN
86	)
87	column_name: str = accessor_domain_kwargs["column"]	×
88	distinct_values: List[pyspark_sql_Row] = (	×
89	df.select(F.col(column_name))
90	.distinct()
91	.where(F.col(column_name).isNotNull())
92	.rdd.flatMap(lambda x: x)
93	.collect()
94	)
95	return set(distinct_values)	×
96
97
98	class ColumnDistinctValuesCount(ColumnAggregateMetricProvider):	3✔
99	metric_name = "column.distinct_values.count"	3✔
100
101	@column_aggregate_value(engine=PandasExecutionEngine)	3✔
102	def _pandas(cls, column: pd.Series, **kwargs) -> int:	3✔
103	return column.nunique()	×
104
105	@column_aggregate_partial(engine=SqlAlchemyExecutionEngine)	3✔
106	def _sqlalchemy(	3✔
107	cls,
108	column: sa_sql_expression_ColumnClause,
109	**kwargs,
110	) -> sa_func_count:
111	"""
112	Past implementations of column.distinct_values.count depended on column.value_counts and column.distinct_values.
113	This was causing performance issues due to the complex query used in column.value_counts and subsequent
114	in-memory operations.
115	"""
116	return sa.func.count(sa.distinct(column))	×
117
118	@column_aggregate_partial(engine=SparkDFExecutionEngine)	3✔
119	def _spark(	3✔
120	cls,
121	column: pyspark_sql_Column,
122	**kwargs,
123	) -> pyspark_sql_Column:
124	"""
125	Past implementations of column.distinct_values.count depended on column.value_counts and column.distinct_values.
126	This was causing performance issues due to the complex query used in column.value_counts and subsequent
127	in-memory operations.
128	"""
129	return F.countDistinct(column)	×
130
131
132	class ColumnDistinctValuesCountUnderThreshold(ColumnAggregateMetricProvider):	3✔
133	metric_name = "column.distinct_values.count.under_threshold"	3✔
134	condition_keys = ("threshold",)	3✔
135
136	@column_aggregate_value(engine=PandasExecutionEngine)	3✔
137	def _pandas(cls, column: pd.Series, threshold: int, **kwargs) -> bool:	3✔
138	return column.nunique() < threshold	×
139
140	@metric_value(engine=SqlAlchemyExecutionEngine)	3✔
141	def _sqlalchemy(	3✔
142	cls,
143	metric_value_kwargs: Dict[str, int],
144	metrics: Dict[str, int],
145	**kwargs,
146	) -> bool:
147	return (	×
148	metrics["column.distinct_values.count"] < metric_value_kwargs["threshold"]
149	)
150
151	@metric_value(engine=SparkDFExecutionEngine)	3✔
152	def _spark(	3✔
153	cls,
154	metric_value_kwargs: Dict[str, int],
155	metrics: Dict[str, int],
156	**kwargs,
157	) -> bool:
158	return (	×
159	metrics["column.distinct_values.count"] < metric_value_kwargs["threshold"]
160	)
161
162	@classmethod	3✔
163	def _get_evaluation_dependencies(	3✔
164	cls,
165	metric: MetricConfiguration,
166	configuration: Optional[ExpectationConfiguration] = None,
167	execution_engine: Optional[ExecutionEngine] = None,
168	runtime_configuration: Optional[Dict] = None,
169	):
170	"""Returns a dictionary of given metric names and their corresponding configuration,
171	specifying the metric types and their respective domains"""
172	dependencies: dict = super()._get_evaluation_dependencies(	×
173	metric=metric,
174	configuration=configuration,
175	execution_engine=execution_engine,
176	runtime_configuration=runtime_configuration,
177	)
178	if metric.metric_name == "column.distinct_values.count.under_threshold":	×
179	dependencies["column.distinct_values.count"] = MetricConfiguration(	×
180	metric_name="column.distinct_values.count",
181	metric_domain_kwargs=metric.metric_domain_kwargs,
182	metric_value_kwargs=None,
183	)
184	return dependencies	×

great-expectations / great_expectations / 20220912.22

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous