[
https://issues.apache.org/jira/browse/SPARK-18473?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
peay updated SPARK-18473:
-------------------------
Description:
I have stumbled onto a corner case where an INNER join appears to return
incorrect results. I believe the join should behave as the identity, but
instead, some values are shuffled around, and some are just plain wrong.
This can be reproduced as follows: joining
{code}
+-----+---------+------+--------+--------+----------+------+
|index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|sessId|
+-----+---------+------+--------+--------+----------+------+
| 1| 1| 1| 0| 1| 0| 1|
| 2| 2| 0| 0| 1| 0| 1|
| 1| 3| 1| 0| 2| 0| 2|
+-----+---------+------+--------+--------+----------+------+
{code}
with
{code}
+------+
|sessId|
+------+
| 1|
| 2|
+------+
{code}
The result is
{code}
+------+-----+---------+------+--------+--------+----------+
|sessId|index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|
+------+-----+---------+------+--------+--------+----------+
| 1| 2| 2| 0| 0| 1| 0|
| 2| 1| 1| 1| 0| 1| -1|
| 2| 1| 3| 1| 0| 2| 0|
+------+-----+---------+------+--------+--------+----------+
{code}
Note how two rows have a sessId of 2 (instead of one row as expected), and how
`fiftyCount` can now be negative while always zero in the original dataframe.
The first dataframe uses two windows:
- `hasOne` uses a `window.rowsBetween(-10, 0)`.
- `hasFifty` uses a `window.rowsBetween(-10, -1)`.
The result is *correct* if:
- `hasFifty` is changed to `window.rowsBetween(-10, 0)` instead of
`window.rowsBetween(-10, -1)`.
- I add {code}.fillna({ 'numOnesBefore': 0 }) {code} after the other call to
`fillna` -- although there are no visible effect on the dataframe as shown by
`show` as far as I can tell.
- I use a LEFT OUTER join instead of INNER JOIN.
- I write both dataframes to Parquet, read them back and join these.
This can be reproduced in pyspark using:
{code}
import pyspark.sql.functions as F
from pyspark.sql.functions import col
from pyspark.sql.window import Window
df1 = sql_context.createDataFrame(
pd.DataFrame({"index": [1, 2, 1], "timeStamp": [1, 2, 3]})
)
window = Window.partitionBy(F.lit(1)).orderBy("timeStamp", "index")
df2 = (
df1
.withColumn("hasOne", (col("index") == 1).cast("int"))
.withColumn("hasFifty", (col("index") == 50).cast("int"))
.withColumn("numOnesBefore",
F.sum(col("hasOne")).over(window.rowsBetween(-10, 0)))
.withColumn("numFiftyStrictlyBefore",
F.sum(col("hasFifty")).over(window.rowsBetween(-10, -1)))
.fillna({ 'numFiftyStrictlyBefore': 0 })
.withColumn("sessId", col("numOnesBefore") - col("numFiftyStrictlyBefore"))
)
df_selector = sql_context.createDataFrame(pd.DataFrame({"sessId": [1, 2]}))
df_joined = df_selector.join(df2, "sessId", how="inner")
df2.show()
df_selector.show()
df_joined.show()
{code}
was:
I have stumbled onto a corner case where an INNER join appears to return
incorrect results. I believe the join should behave as the identity, but
instead, some values are shuffled around, and some are just plain wrong.
This can be reproduced as follows: joining
{code}
+-----+---------+------+--------+--------+----------+------+
|index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|sessId|
+-----+---------+------+--------+--------+----------+------+
| 1| 1| 1| 0| 1| 0| 1|
| 2| 2| 0| 0| 1| 0| 1|
| 1| 3| 1| 0| 2| 0| 2|
+-----+---------+------+--------+--------+----------+------+
{code}
with
{code}
+------+
|sessId|
+------+
| 1|
| 2|
+------+
{code}
The result is
{code}
+------+-----+---------+------+--------+--------+----------+
|sessId|index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|
+------+-----+---------+------+--------+--------+----------+
| 1| 2| 2| 0| 0| 1| 0|
| 2| 1| 1| 1| 0| 1| -1|
| 2| 1| 3| 1| 0| 2| 0|
+------+-----+---------+------+--------+--------+----------+
{code}
Note how rows have a sessId of 2 (instead of one row as expected), and how
`fiftyCount` can now be negative while always zero in the original dataframe.
The first dataframe uses two windows:
- `hasOne` uses a `window.rowsBetween(-10, 0)`.
- `hasFifty` uses a `window.rowsBetween(-10, -1)`.
The result is *correct* if:
- `hasFifty` is changed to `window.rowsBetween(-10, 0)` instead of
`window.rowsBetween(-10, -1)`.
- I add {code}.fillna({ 'numOnesBefore': 0 }) {code} after the other call to
`fillna` -- although there are no visible effect on the dataframe as shown by
`show` as far as I can tell.
- I use a LEFT OUTER join instead of INNER JOIN.
- I write both dataframes to Parquet, read them back and join these.
This can be reproduced in pyspark using:
{code}
import pyspark.sql.functions as F
from pyspark.sql.functions import col
from pyspark.sql.window import Window
df1 = sql_context.createDataFrame(
pd.DataFrame({"index": [1, 2, 1], "timeStamp": [1, 2, 3]})
)
window = Window.partitionBy(F.lit(1)).orderBy("timeStamp", "index")
df2 = (
df1
.withColumn("hasOne", (col("index") == 1).cast("int"))
.withColumn("hasFifty", (col("index") == 50).cast("int"))
.withColumn("numOnesBefore",
F.sum(col("hasOne")).over(window.rowsBetween(-10, 0)))
.withColumn("numFiftyStrictlyBefore",
F.sum(col("hasFifty")).over(window.rowsBetween(-10, -1)))
.fillna({ 'numFiftyStrictlyBefore': 0 })
.withColumn("sessId", col("numOnesBefore") - col("numFiftyStrictlyBefore"))
)
df_selector = sql_context.createDataFrame(pd.DataFrame({"sessId": [1, 2]}))
df_joined = df_selector.join(df2, "sessId", how="inner")
df2.show()
df_selector.show()
df_joined.show()
{code}
> Correctness issue in INNER join result with window functions
> ------------------------------------------------------------
>
> Key: SPARK-18473
> URL: https://issues.apache.org/jira/browse/SPARK-18473
> Project: Spark
> Issue Type: Bug
> Components: PySpark, Spark Core, SQL
> Affects Versions: 2.0.1
> Reporter: peay
>
> I have stumbled onto a corner case where an INNER join appears to return
> incorrect results. I believe the join should behave as the identity, but
> instead, some values are shuffled around, and some are just plain wrong.
> This can be reproduced as follows: joining
> {code}
> +-----+---------+------+--------+--------+----------+------+
> |index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|sessId|
> +-----+---------+------+--------+--------+----------+------+
> | 1| 1| 1| 0| 1| 0| 1|
> | 2| 2| 0| 0| 1| 0| 1|
> | 1| 3| 1| 0| 2| 0| 2|
> +-----+---------+------+--------+--------+----------+------+
> {code}
> with
> {code}
> +------+
> |sessId|
> +------+
> | 1|
> | 2|
> +------+
> {code}
> The result is
> {code}
> +------+-----+---------+------+--------+--------+----------+
> |sessId|index|timeStamp|hasOne|hasFifty|oneCount|fiftyCount|
> +------+-----+---------+------+--------+--------+----------+
> | 1| 2| 2| 0| 0| 1| 0|
> | 2| 1| 1| 1| 0| 1| -1|
> | 2| 1| 3| 1| 0| 2| 0|
> +------+-----+---------+------+--------+--------+----------+
> {code}
> Note how two rows have a sessId of 2 (instead of one row as expected), and
> how `fiftyCount` can now be negative while always zero in the original
> dataframe.
> The first dataframe uses two windows:
> - `hasOne` uses a `window.rowsBetween(-10, 0)`.
> - `hasFifty` uses a `window.rowsBetween(-10, -1)`.
> The result is *correct* if:
> - `hasFifty` is changed to `window.rowsBetween(-10, 0)` instead of
> `window.rowsBetween(-10, -1)`.
> - I add {code}.fillna({ 'numOnesBefore': 0 }) {code} after the other call to
> `fillna` -- although there are no visible effect on the dataframe as shown by
> `show` as far as I can tell.
> - I use a LEFT OUTER join instead of INNER JOIN.
> - I write both dataframes to Parquet, read them back and join these.
> This can be reproduced in pyspark using:
> {code}
> import pyspark.sql.functions as F
> from pyspark.sql.functions import col
> from pyspark.sql.window import Window
> df1 = sql_context.createDataFrame(
> pd.DataFrame({"index": [1, 2, 1], "timeStamp": [1, 2, 3]})
> )
> window = Window.partitionBy(F.lit(1)).orderBy("timeStamp", "index")
> df2 = (
> df1
> .withColumn("hasOne", (col("index") == 1).cast("int"))
> .withColumn("hasFifty", (col("index") == 50).cast("int"))
> .withColumn("numOnesBefore",
> F.sum(col("hasOne")).over(window.rowsBetween(-10, 0)))
> .withColumn("numFiftyStrictlyBefore",
> F.sum(col("hasFifty")).over(window.rowsBetween(-10, -1)))
> .fillna({ 'numFiftyStrictlyBefore': 0 })
> .withColumn("sessId", col("numOnesBefore") -
> col("numFiftyStrictlyBefore"))
> )
> df_selector = sql_context.createDataFrame(pd.DataFrame({"sessId": [1, 2]}))
> df_joined = df_selector.join(df2, "sessId", how="inner")
> df2.show()
> df_selector.show()
> df_joined.show()
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
