You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "devin-petersohn (via GitHub)" <gi...@apache.org> on 2026/04/01 18:20:34 UTC
[PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
devin-petersohn opened a new pull request, #55143:
URL: https://github.com/apache/spark/pull/55143
### What changes were proposed in this pull request?
Implement `DataFrame.compare` for the pandas API on Spark. This was already implemented for `Series` but was unimplemented for `DataFrame`.
### Why are the changes needed?
Implements missing API to improve pandas compatibility.
### Does this PR introduce _any_ user-facing change?
Yes, new `DataFrame.compare` method.
### How was this patch tested?
CI
### Was this patch authored or co-authored using generative AI tooling?
Co-authored-by: Claude Opus 4
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
Posted by "devin-petersohn (via GitHub)" <gi...@apache.org>.
devin-petersohn commented on PR #55143:
URL: https://github.com/apache/spark/pull/55143#issuecomment-4201836772
@zhengruifeng, @gaogaotiantian, or @Yicong-Huang - any thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #55143:
URL: https://github.com/apache/spark/pull/55143#discussion_r3047963415
##########
python/pyspark/pandas/tests/diff_frames_ops/test_compare_frame.py:
##########
@@ -0,0 +1,195 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import numpy as np
+import pandas as pd
+
+from pyspark import pandas as ps
+from pyspark.pandas.config import set_option, reset_option
+from pyspark.testing.pandasutils import PandasOnSparkTestCase
+from pyspark.testing.sqlutils import SQLTestUtils
+
+
+class CompareFrameMixin:
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ set_option("compute.ops_on_diff_frames", True)
+
+ @classmethod
+ def tearDownClass(cls):
+ reset_option("compute.ops_on_diff_frames")
+ super().tearDownClass()
+
+ def _assert_compare_eq(self, pdf1, pdf2, **kwargs):
+ """Helper to compare pandas and pyspark compare results.
+
+ Uses almost=True when keep_equal is False (the default) because
+ pandas promotes integer columns to float when NaN masking is applied
+ (before row filtering), while pyspark applies row filtering first, so
+ columns that end up with no NaN values may keep their original dtype.
+ """
+ psdf1 = ps.from_pandas(pdf1)
+ psdf2 = ps.from_pandas(pdf2)
+ almost = not kwargs.get("keep_equal", False)
+ self.assert_eq(
+ pdf1.compare(pdf2, **kwargs).sort_index(),
+ psdf1.compare(psdf2, **kwargs).sort_index(),
+ almost=almost,
+ )
+
+ def test_compare(self):
+ pdf1 = pd.DataFrame({"a": ["b", "c", np.nan, "g", np.nan], "b": [1, 2, 3, 4, np.nan]})
+ pdf2 = pd.DataFrame({"a": ["a", "c", np.nan, np.nan, "h"], "b": [1, 2, 1, 4, np.nan]})
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_equal=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True, keep_equal=True)
+
+ def test_compare_same_anchor(self):
+ psdf = ps.DataFrame({"a": [1, 2, 3, 4, 5], "b": [5, 4, 3, 2, 1]})
+ psdf1 = psdf[["a", "b"]]
+ psdf2 = psdf[["a", "b"]].rename(columns={"a": "a", "b": "b"})
+
+ pdf = psdf.to_pandas()
+ pdf1 = pdf[["a", "b"]]
+ pdf2 = pdf[["a", "b"]]
+
+ self.assert_eq(
+ pdf1.compare(pdf2).sort_index(),
+ psdf1.compare(psdf2).sort_index(),
+ )
+ self.assert_eq(
+ pdf1.compare(pdf2, keep_shape=True).sort_index(),
+ psdf1.compare(psdf2, keep_shape=True).sort_index(),
+ )
+ self.assert_eq(
+ pdf1.compare(pdf2, keep_equal=True).sort_index(),
+ psdf1.compare(psdf2, keep_equal=True).sort_index(),
+ )
+ self.assert_eq(
+ pdf1.compare(pdf2, keep_shape=True, keep_equal=True).sort_index(),
+ psdf1.compare(psdf2, keep_shape=True, keep_equal=True).sort_index(),
+ )
+
+ def test_compare_multiindex(self):
+ pdf1 = pd.DataFrame({"a": ["b", "c", np.nan, "g", np.nan], "b": [1, 2, 3, 4, np.nan]})
+ pdf2 = pd.DataFrame({"a": ["a", "c", np.nan, np.nan, "h"], "b": [1, 2, 1, 4, np.nan]})
+ midx = pd.MultiIndex.from_tuples(
+ [("a", "x"), ("b", "y"), ("c", "z"), ("x", "k"), ("q", "l")]
+ )
+ pdf1.index = midx
+ pdf2.index = midx
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_equal=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True, keep_equal=True)
+
+ def test_compare_all_equal(self):
+ pdf1 = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
+ pdf2 = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_equal=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True, keep_equal=True)
+
+ def test_compare_all_different(self):
+ pdf1 = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
+ pdf2 = pd.DataFrame({"a": [9, 8, 7], "b": [6, 5, 4]})
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_equal=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True, keep_equal=True)
+
+ def test_compare_nans_matching(self):
+ pdf1 = pd.DataFrame({"a": [np.nan, np.nan, 1.0], "b": [np.nan, 2.0, np.nan]})
+ pdf2 = pd.DataFrame({"a": [np.nan, np.nan, 1.0], "b": [np.nan, 2.0, np.nan]})
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+
+ def test_compare_nans_different(self):
+ pdf1 = pd.DataFrame({"a": [np.nan, 1.0], "b": [2.0, np.nan]})
+ pdf2 = pd.DataFrame({"a": [1.0, np.nan], "b": [np.nan, 2.0]})
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_equal=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True, keep_equal=True)
+
+ def test_compare_single_column(self):
+ pdf1 = pd.DataFrame({"a": [1, 2, 3]})
+ pdf2 = pd.DataFrame({"a": [1, 9, 3]})
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_equal=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True, keep_equal=True)
+
+ def test_compare_empty(self):
+ pdf1 = pd.DataFrame({"a": pd.Series([], dtype="int64"), "b": pd.Series([], dtype="int64")})
+ pdf2 = pd.DataFrame({"a": pd.Series([], dtype="int64"), "b": pd.Series([], dtype="int64")})
+ self._assert_compare_eq(pdf1, pdf2)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_equal=True)
+ self._assert_compare_eq(pdf1, pdf2, keep_shape=True, keep_equal=True)
+
+ def test_compare_different_index(self):
+ with self.assertRaisesRegex(
+ ValueError, "Can only compare identically-labeled DataFrame objects"
+ ):
+ psdf1 = ps.DataFrame(
+ {"a": [1, 2, 3, 4, 5], "b": [1, 2, 3, 4, 5]},
+ index=pd.Index([1, 2, 3, 4, 5]),
+ )
+ psdf2 = ps.DataFrame(
+ {"a": [2, 2, 3, 4, 1], "b": [2, 2, 3, 4, 1]},
+ index=pd.Index([5, 4, 3, 2, 1]),
+ )
+ psdf1.compare(psdf2)
Review Comment:
nit: Can we move the data initialization out of the `assertRaisesRegex` block? I think having the line that would actually raise in the block only will make it easier to understand.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #55143:
URL: https://github.com/apache/spark/pull/55143#issuecomment-5075901356
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #55143: [SPARK-40328][PS] Implement DataFrame.compare
URL: https://github.com/apache/spark/pull/55143
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #55143:
URL: https://github.com/apache/spark/pull/55143#issuecomment-4189659673
Seems fine. cc @zhengruifeng @gaogaotiantian @Yicong-Huang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
Posted by "gaogaotiantian (via GitHub)" <gi...@apache.org>.
gaogaotiantian commented on code in PR #55143:
URL: https://github.com/apache/spark/pull/55143#discussion_r3047970394
##########
python/pyspark/pandas/frame.py:
##########
@@ -9233,6 +9233,186 @@ def combine_first(self, other: "DataFrame") -> "DataFrame":
)
return DataFrame(internal)
+ def compare(
+ self, other: "DataFrame", keep_shape: bool = False, keep_equal: bool = False
Review Comment:
We have some historical burdens about having different parameters for our pandas-like API, which could be confusing when they are used with positional arguments. However, for new APIs, how about we make `keep_shape` and `keep_equal` keyword only? If we want to add `align_axis` in the future, that won't be a problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-40328][PS] Implement DataFrame.compare [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #55143:
URL: https://github.com/apache/spark/pull/55143#discussion_r3048185880
##########
python/pyspark/pandas/frame.py:
##########
@@ -9233,6 +9233,186 @@ def combine_first(self, other: "DataFrame") -> "DataFrame":
)
return DataFrame(internal)
+ def compare(
+ self, other: "DataFrame", keep_shape: bool = False, keep_equal: bool = False
+ ) -> "DataFrame":
+ """
+ Compare to another DataFrame and show the differences.
+
+ .. versionadded:: 4.2.0
+
+ .. note:: This API is slightly different from pandas when indexes from both
+ DataFrames are not identical and config 'compute.eager_check' is False.
+ pandas raises an exception; however, pandas-on-Spark just proceeds and
+ performs by ignoring mismatches.
+
+ Parameters
+ ----------
+ other : DataFrame
+ Object to compare with.
+ keep_shape : bool, default False
+ If true, all rows and columns are kept.
+ Otherwise, only the ones with different values are kept.
+ keep_equal : bool, default False
+ If true, the result keeps values that are equal.
+ Otherwise, equal values are shown as NaNs.
+
+ Returns
+ -------
+ DataFrame
+ DataFrame that shows the differences stacked side by side.
+ The resulting index will be a MultiIndex with 'self' and 'other'
+ stacked alternately at the inner level.
+
+ Notes
+ -----
+ Matching NaNs will not appear as a difference.
+
+ Examples
+ --------
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> df1 = ps.DataFrame(
+ ... {"col1": [1.0, 2.0, 3.0, 4.0, 5.0],
+ ... "col2": [1.0, 2.0, 3.0, np.nan, 5.0],
+ ... "col3": [1.0, 2.0, 3.0, 4.0, 5.0]},
+ ... columns=["col1", "col2", "col3"],
+ ... )
+ >>> df2 = ps.DataFrame(
+ ... {"col1": [9.0, 2.0, 3.0, 4.0, 5.0],
+ ... "col2": [1.0, 2.0, 1.0, np.nan, 5.0],
+ ... "col3": [1.0, 2.0, 3.0, 4.0, 5.0]},
+ ... columns=["col1", "col2", "col3"],
+ ... )
+
+ Align the differences on columns
+
+ >>> df1.compare(df2).sort_index() # doctest: +NORMALIZE_WHITESPACE
+ col1 col2
+ self other self other
+ 0 1.0 9.0 NaN NaN
+ 2 NaN NaN 3.0 1.0
+
+ Keep all original rows
+
+ >>> df1.compare(df2, keep_shape=True).sort_index() # doctest: +NORMALIZE_WHITESPACE
+ col1 col2 col3
+ self other self other self other
+ 0 1.0 9.0 NaN NaN NaN NaN
+ 1 NaN NaN NaN NaN NaN NaN
+ 2 NaN NaN 3.0 1.0 NaN NaN
+ 3 NaN NaN NaN NaN NaN NaN
+ 4 NaN NaN NaN NaN NaN NaN
+
+ Keep all original rows and all original values
+
+ >>> df1.compare(df2, keep_shape=True, keep_equal=True).sort_index() # doctest: +NORMALIZE_WHITESPACE
+ col1 col2 col3
+ self other self other self other
+ 0 1.0 9.0 1.0 1.0 1.0 1.0
+ 1 2.0 2.0 2.0 2.0 2.0 2.0
+ 2 3.0 3.0 3.0 1.0 3.0 3.0
+ 3 4.0 4.0 NaN NaN 4.0 4.0
+ 4 5.0 5.0 5.0 5.0 5.0 5.0
+
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, DataFrame):
+ raise TypeError("`compare` only allows `DataFrame` for parameter `other`")
+ if not self.columns.equals(other.columns):
+ raise ValueError("can only compare identically-labeled DataFrame objects")
+
+ if same_anchor(self, other):
+ combined = self
+ this = self
+ that = other
+ else:
+ if get_option("compute.eager_check") and not self.index.equals(other.index):
+ raise ValueError("Can only compare identically-labeled DataFrame objects")
+ combined = combine_frames(self, other)
+ this = combined["this"]
+ that = combined["that"]
+
+ column_labels: List[Tuple] = []
+ data_spark_columns = []
+ data_col_names: List[str] = []
+ diff_conditions: List[Tuple[Tuple, PySparkColumn]] = []
+ index_scols = combined._internal.index_spark_columns
+ sdf = combined._internal.spark_frame
+
+ for column_label in self._internal.column_labels:
+ this_scol = this._internal.spark_column_for(column_label)
+ that_scol = that._internal.spark_column_for(column_label)
+
+ this_col_name = name_like_string(column_label) + "_self"
+ that_col_name = name_like_string(column_label) + "_other"
+ is_equal = this_scol.eqNullSafe(that_scol)
+
+ if not keep_shape:
+ diff_conditions.append((column_label, ~is_equal))
+
+ if keep_equal:
+ data_spark_columns.append(this_scol.alias(this_col_name))
+ data_spark_columns.append(that_scol.alias(that_col_name))
+ else:
+ # Null-mask equal values so differences are visible.
+ this_masked = F.when(is_equal, None).otherwise(this_scol)
+ that_masked = F.when(is_equal, None).otherwise(that_scol)
+ data_spark_columns.append(this_masked.alias(this_col_name))
+ data_spark_columns.append(that_masked.alias(that_col_name))
+
+ data_col_names.extend([this_col_name, that_col_name])
+ column_labels.append(column_label + ("self",))
+ column_labels.append(column_label + ("other",))
+
+ if diff_conditions:
+ # Filter rows where at least one column differs.
+ row_mask = reduce(lambda a, b: a | b, [cond for _, cond in diff_conditions])
+ sdf = sdf.filter(row_mask)
+
+ # Determine which columns have any difference and prune the rest.
+ has_diff = sdf.select(
+ [
+ F.max(cond.cast("int")).alias(name_like_string(label))
+ for label, cond in diff_conditions
+ ]
+ ).head()
Review Comment:
There seems to be two data passes/data scans: the first one to determine which columns have any differences, and the second pass to filter the data. is it possible to reduce the passes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org