Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[MAINTENANCE] Performance improvement refactor for Spark unexpected values #3368

Merged
merged 32 commits into from Sep 21, 2021
Merged

[MAINTENANCE] Performance improvement refactor for Spark unexpected values #3368

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
32 commits
Select commit Hold shift + click to select a range
ec72b07
[MAINTENANCE] Performance improvement refactor for helper _spark_colu…
Sep 8, 2021
5ad62bc
Merge branch 'develop' into working-branch/DEVREL-154/improve_SparkDF…
NathanFarmer Sep 8, 2021
778051e
Merge branch 'develop' into working-branch/DEVREL-154/improve_SparkDF…
NathanFarmer Sep 16, 2021
4473092
[MAINTENANCE] Performance improvement refactor for helper _spark_colu…
Sep 16, 2021
a784897
Merge branch 'working-branch/DEVREL-154/improve_SparkDFExecutionEngin…
Sep 16, 2021
66b5ae3
Change log
Sep 16, 2021
21eea41
[MAINTENANCE] This test no longer applies to spark because we stopped…
Sep 17, 2021
a39aa19
[MAINTENANCE] Remove all sorting logic from spark provider helpers (#…
Sep 17, 2021
abf47b8
[MAINTENANCE] Sort dictionaries in tests for comparisons (#3368).
Sep 17, 2021
a276eab
Merge branch 'develop' into working-branch/DEVREL-154/improve_SparkDF…
NathanFarmer Sep 17, 2021
fea4e71
Linting
Sep 17, 2021
bd8fc07
Merge branch 'working-branch/DEVREL-154/improve_SparkDFExecutionEngin…
Sep 17, 2021
8050ce9
Clean up
Sep 17, 2021
522a3f3
[MAINTENANCE] Incorrect source of __lt__ in comment (#3368).
Sep 17, 2021
fcb22b8
[MAINTENANCE] Clarify how sorting works for each data type (#3368).
Sep 17, 2021
dd56b24
[MAINTENANCE] Lambda instead of itemgetter for consistency/simplicity…
Sep 17, 2021
ce2105e
Linting
Sep 17, 2021
8b48961
Accidentally re-used variable name
Sep 17, 2021
2e2ff25
Linting
Sep 17, 2021
7a72aaa
[MAINTENANCE] Change final use of boolean_mapped_unexpected_values to…
Sep 21, 2021
19f3bd4
[MAINTENANCE] Helper function for sorting unexpected_values during te…
Sep 21, 2021
7adf3b9
[MAINTENANCE] When exact_match_out is True we still need to sort unex…
Sep 21, 2021
b9bb7cf
[MAINTENANCE] Moved sort logic into helper function (#3368).
Sep 21, 2021
099802a
Cleanup
Sep 21, 2021
1f9d8d3
[MAINTENANCE] Sort should also be applied to partial_unexpected_list …
Sep 21, 2021
93b5d25
[MAINTENANCE] Revert broken test back to its original state (#3368).
Sep 21, 2021
80074e9
Linting
Sep 21, 2021
e1ddfee
Merge branch 'develop' into working-branch/DEVREL-154/improve_SparkDF…
NathanFarmer Sep 21, 2021
341b769
[MAINTENANCE] Consolidate sorting to make it clear that we do it whet…
Sep 21, 2021
d338d03
Merge branch 'working-branch/DEVREL-154/improve_SparkDFExecutionEngin…
Sep 21, 2021
51f2005
Linting
Sep 21, 2021
753e44d
Merge branch 'develop' into working-branch/DEVREL-154/improve_SparkDF…
NathanFarmer Sep 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs_rtd/changelog.rst
View file
Open in desktop
Expand Up @@ -7,6 +7,7 @@ Changelog

develop
-----------------
* [MAINTENANCE] Spark performance improvement for metrics that return unexpected values (#3368)

0.13.34
-----------------
Expand Down Expand Up @@ -64,7 +65,6 @@ develop
* [MAINTENANCE] Tests for RuntimeDataConnector at Datasource-level (Spark and Pandas) (#3318)
* [MAINTENANCE] Various doc patches (#3326)
* [MAINTENANCE] clean up imports and method signatures (#3337)
>>>>>>> 9208de453238af6d673aa9184c865b8422165172

0.13.31
-----------------
Expand Down
61 changes: 15 additions & 46 deletions great_expectations/expectations/metrics/map_metric_provider.py
View file
Open in desktop
Expand Up @@ -2338,6 +2338,7 @@ def _spark_map_condition_unexpected_count_value(
df = execution_engine.get_domain_records(
domain_kwargs=domain_kwargs,
)
# withColumn is required to transform window functions returned by some metrics to boolean mask
data = df.withColumn("__unexpected", unexpected_condition)
filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))
return filtered.count()
Expand Down Expand Up @@ -2373,17 +2374,9 @@ def _spark_column_map_condition_values(
message=f'Error: The column "{column_name}" in BatchData does not exist.'
)

data = (
df.withColumn("__row_number", F.row_number().over(Window.orderBy(F.lit(1))))
.withColumn("__unexpected", unexpected_condition)
.orderBy(F.col("__row_number"))
)

filtered = (
data.filter(F.col("__unexpected") == True)
.drop(F.col("__unexpected"))
.drop(F.col("__row_number"))
)
# withColumn is required to transform window functions returned by some metrics to boolean mask
data = df.withColumn("__unexpected", unexpected_condition)
filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))

result_format = metric_value_kwargs["result_format"]
if result_format["result_format"] == "COMPLETE":
Expand Down Expand Up @@ -2411,7 +2404,10 @@ def _spark_column_map_condition_value_counts(
df = execution_engine.get_domain_records(
domain_kwargs=compute_domain_kwargs,
)

# withColumn is required to transform window functions returned by some metrics to boolean mask
data = df.withColumn("__unexpected", unexpected_condition)
filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))

if "column" not in accessor_domain_kwargs:
raise ValueError(
Expand All @@ -2429,7 +2425,6 @@ def _spark_column_map_condition_value_counts(

result_format = metric_value_kwargs["result_format"]

filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer I believe that we can leave the filtered = statement where it was (because there could be an exception raised earlier, so no need to have it before). Would you agree or not? Thanks.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alexsherstinsky Sure, we can leave this line where it is, but then I would advocate to move lines 2408-2409 down below as well for consistency.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer The way I am seeing it now seems good. Thank you.

value_counts = filtered.groupBy(F.col(column_name)).count()
if result_format["result_format"] == "COMPLETE":
rows = value_counts.collect()
Expand Down Expand Up @@ -2458,17 +2453,9 @@ def _spark_map_condition_rows(
domain_kwargs=domain_kwargs,
)

data = (
df.withColumn("__row_number", F.row_number().over(Window.orderBy(F.lit(1))))
.withColumn("__unexpected", unexpected_condition)
.orderBy(F.col("__row_number"))
)

filtered = (
data.filter(F.col("__unexpected") == True)
.drop(F.col("__unexpected"))
.drop(F.col("__row_number"))
)
# withColumn is required to transform window functions returned by some metrics to boolean mask
data = df.withColumn("__unexpected", unexpected_condition)
filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))

result_format = metric_value_kwargs["result_format"]

Expand Down Expand Up @@ -2514,17 +2501,7 @@ def _spark_column_pair_map_condition_values(
message=f'Error: The column "{column_name}" in BatchData does not exist.'
)

data = (
df.withColumn("__row_number", F.row_number().over(Window.orderBy(F.lit(1))))
.withColumn("__unexpected", boolean_mapped_unexpected_values)
.orderBy(F.col("__row_number"))
)

filtered = (
data.filter(F.col("__unexpected") == True)
.drop(F.col("__unexpected"))
.drop(F.col("__row_number"))
)
filtered = df.filter(boolean_mapped_unexpected_values)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer Can we please preserve the pattern for having data = df.withColumn("__unexpected", boolean_mapped_unexpected_values) first -- and then followed by the filter on F.col("__unexpected") -- for consistency and readability purposes (plus for potential extensibility needs). Thank you!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

See my other comments about that pattern being measurably slower. We can discuss the tradeoffs in our face-to-face testing discussion.


result_format = metric_value_kwargs["result_format"]
if result_format["result_format"] == "COMPLETE":
Expand Down Expand Up @@ -2585,7 +2562,7 @@ def _spark_multicolumn_map_condition_values(
):
"""Return values from the specified domain that match the map-style metric in the metrics dictionary."""
(
boolean_mapped_unexpected_values,
unexpected_condition,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer Could we place keep this variable named boolean_mapped_unexpected_values as it was before (I was following the previous style in such methods across all execution engines and would like to keep it consistent, unless you have a strong reason for changing it now). Thanks!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The reason for changing this was that metrics are not returning booleans in all cases. Cases that return a window function would fail for the 1-line solution filtered = df.filter(boolean_mapped_unexpected_values). The 2-line solution using withColumn creates a new boolean mapped column from this variable. You can see how I left it alone on line 2504 because it actually is a boolean mapping for all cases there.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer I cannot find all the line numbers involved (the UI here is confusing, but I follow your logic). Thank you! P.S.: Should we standardize all cases to use unexpected_condition? Or do we want to preserve a special spot and variable name for the situations where it will always be strictly-boolean mapped? Thoughts welcome. Thank you.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alexsherstinsky If we were to standardize all cases to use unexpected_condition that would future-proof metrics that return a window function. The tradeoff is that using boolean_mapped_unexpected_values with df.filter (1-line solution) is measurably faster where it is possible. I would propose that we leave boolean_mapped_unexpected_values as-is in places where it is not currently needed, and if a new metric requires it to be changed, it can be done at that time.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer Perhaps we need to clean up our code? Can you please help us out here -- I see in Pandas, SQL, and Spark the pattern that in some cases defines unexpected_condition and in other cases defines boolean_mapped_unexpected_values as variable names, but ultimately these variables are used the same way: in a WHERE type clause. So are we simply misnaming this variable in one case for spark, because it is truly an (unexpected) condition and not merely boolean-mapped (unexpected) values? If this is correct, then please go ahead and fix as you deem appropriate. Thanks so much!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alexsherstinsky It is correct that the metric always returns an unexpected_condition for Spark. So for semantic correctness, I have changed the one remaining use of Spark boolean_mapped_unexpected_values to unexpected_condition. In most Spark cases unexpected_condition can be passed directly to a WHERE clause, except in the case of a window function. I went ahead and changed:

    data = df.filter(unexpected_condition)

to:

    data = df.withColumn("__unexpected", unexpected_condition)
    filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))

for consistency in this single case.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer Thanks -- at least it is consistent, and we can revise later. We now need to solve the remaining problem: the equivalency of results, when the sort order is not enforced. If we can do this, then it is a big gain!

compute_domain_kwargs,
accessor_domain_kwargs,
) = metrics["unexpected_condition"]
Expand Down Expand Up @@ -2613,17 +2590,9 @@ def _spark_multicolumn_map_condition_values(
message=f'Error: The column "{column_name}" in BatchData does not exist.'
)

data = (
df.withColumn("__row_number", F.row_number().over(Window.orderBy(F.lit(1))))
.withColumn("__unexpected", boolean_mapped_unexpected_values)
.orderBy(F.col("__row_number"))
)

filtered = (
data.filter(F.col("__unexpected") == True)
.drop(F.col("__unexpected"))
.drop(F.col("__row_number"))
)
# withColumn is required to transform window functions returned by some metrics to boolean mask
data = df.withColumn("__unexpected", unexpected_condition)
filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))

column_selector = [F.col(column_name) for column_name in column_list]

Expand Down
8 changes: 8 additions & 0 deletions great_expectations/self_check/util.py
View file
Open in desktop
Expand Up @@ -10,6 +10,7 @@
from functools import wraps
from types import ModuleType
from typing import Dict, List, Optional, Union
from operator import itemgetter

import numpy as np
import pandas as pd
Expand Down Expand Up @@ -1930,6 +1931,13 @@ def check_json_test_result(test, result, data_asset=None):
elif key == "unexpected_list":
# check if value can be sorted; if so, sort so arbitrary ordering of results does not cause failure
if (isinstance(value, list)) & (len(value) >= 1):
# dictionary handling isn't implemented in great_expectations.core.data_context_key.__lt__
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer I ❤️ the idea here; I have two questions about it:

  1. Would it be better to follow up on your comment and update great_expectations.core.data_context_key.__lt_ or not?
  2. If we must do this "consistent sort" here, is there a strong justification for using itemgetter instead of just lambda? If at all possible, I would prefer the lambda (for simplicity).

Thank you! I love how this idea absolves us from the need for the computationally-prohibitive row_number()!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I clarified the lt comment (it is actually looking at the built-in Python class that is passed). I also changed itemgetter to lambda.

# but values still need to be sorted since spark metrics return unordered
if isinstance(value[0], dict):
value = sorted(value, key=itemgetter(*list(value[0].keys())))
result["result"]["unexpected_list"] = sorted(
result["result"]["unexpected_list"], key=itemgetter(*list(value[0].keys()))
)
if type(value[0].__lt__(value[0])) != type(NotImplemented):
value = sorted(value, key=lambda x: str(x))
result["result"]["unexpected_list"] = sorted(
Expand Down
2 changes: 1 addition & 1 deletion ...s/test_definitions/multicolumn_map_expectations/expect_compound_columns_to_be_unique.json
View file
Open in desktop
Expand Up @@ -213,7 +213,7 @@
},{
"title": "unexpected_values_exact_match_out_without_unexpected_index_list",
"exact_match_out" : true,
"suppress_test_for": ["pandas"],
"only_for": ["sqlalchemy"],
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"exact_match_out" : true will no longer work for spark since we are no longer sorting results

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer I feel that we must make it (or an equivalent, additional test work for Spark). The existing test excludes Pandas, because unexpected_index_list only applies to Pandas; however, the functionality must work for the other execution engines. Since you introduced the "force-sorting" of both sides in the assertion in self_check/util.py, why would the existing test not work? To me, having this test is critical. Thank you!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In self_check/util.py#L1882 when test["exact_match_out"] is True all of the test sort logic is foregone. There is no analog to this test that would work that I can think of.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note: There is a test above this one for pandas. It is possible if we sort the data in the test harness it will work for spark but I don't think even that is guaranteed.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@NathanFarmer I feel that we should discuss this in order to find a solution for testing the functionality for the Spark engine. The reasoning is that the addition of row_number() and sorting on it was key to making Spark work. However, I believe that you are pointing out that the sort order of the results is the only difference, the output being the same as expected otherwise. If this is the case, then we need to figure out how to show that in a test. If I am not understanding this correctly, then we might have to disable the expectation for Spark if it is qualitatively wrong without the use of row_number() (and incurring the corresponding performance penalty). Thank you!

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alexsherstinsky I was able to make changes to self_check/util.py to address this and reverted this test back to its original state. Please let me know your thoughts on the code, but the short version is we are now sorting unexpected_values and partial_unexpected_values in both cases where test["exact_match_out"] is True or False on Spark dataframes.

"in": {
"column_list": ["a", "b"]
},
Expand Down