[
https://issues.apache.org/jira/browse/SPARK-20073?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15939883#comment-15939883
]
Everett Anderson commented on SPARK-20073:
------------------------------------------
Ah, interesting.
I notice that you do have to use a different value for the alias -- it
recognizes if it's the same string and does nothing.
This seems like a work-around for a rather surprising "gotcha" in Spark's joins
when everything else works so well.
Do you know if a longer-term fix is being worked on?
Separately, trying to generalize your solution a bit further, I suppose one
could do something like this to rename all columns before and after a join,
though it's rather gross --
{noformat}
// Rename all columns
val renamedPeople = people.select(people.columns.map(x =>
people.col(x).as("tmp" + x)):_*)
val renamedVariantCounts = variantCounts.select(variantCounts.columns.map(x =>
variantCounts.col(x).as("tmp" + x)):_*)
// Perform the join
val joinedResult = renamedVariantCounts.join(renamedPeople,
renamedVariantCounts("tmpname")<=>renamedPeople("tmpname"))
// Restore the original column names
val result = joinedResult.select((renamedVariantCounts.columns.map(x =>
renamedVariantCounts.col(x).as(x.substring(3))) ++ renamedPeople.columns.map(x
=> renamedPeople.col(x).as(x.substring(3)))):_*)
result.show
+----+-------------+----+-----+----+
|name|variant_count|name|group|data|
+----+-------------+----+-----+----+
|Fred| 2|Fred| 8| 1|
|Fred| 2|Fred| 8| 2|
| Amy| 2| Amy| 12| 5|
| Amy| 2| Amy| 12| 6|
|null| 2|null| 10| 3|
|null| 2|null| 10| 4|
+----+-------------+----+-----+----+
result.explain
== Physical Plan ==
*Project [tmpname#30 AS name#54, tmpvariant_count#31L AS variant_count#55L,
tmpname#23 AS name#56, tmpgroup#24 AS group#57, tmpdata#25 AS data#58]
+- *SortMergeJoin [coalesce(tmpname#30, )], [coalesce(tmpname#23, )], Inner,
(tmpname#30 <=> tmpname#23)
:- *Sort [coalesce(tmpname#30, ) ASC], false, 0
: +- Exchange hashpartitioning(coalesce(tmpname#30, ), 200)
: +- *Project [name#3 AS tmpname#30, variant_count#11L AS
tmpvariant_count#31L]
: +- *Filter (variant_count#11L > 1)
: +- *HashAggregate(keys=[name#3], functions=[count(distinct
struct(name#3, group#4, data#5)#86)])
: +- Exchange hashpartitioning(name#3, 200)
: +- *HashAggregate(keys=[name#3],
functions=[partial_count(distinct struct(name#3, group#4, data#5)#86)])
: +- *HashAggregate(keys=[name#3, struct(name#3, group#4,
data#5)#86], functions=[])
: +- Exchange hashpartitioning(name#3, struct(name#3,
group#4, data#5)#86, 200)
: +- *HashAggregate(keys=[name#3, struct(name#3,
group#4, data#5) AS struct(name#3, group#4, data#5)#86], functions=[])
: +- Scan ExistingRDD[name#3,group#4,data#5]
+- *Sort [coalesce(tmpname#23, ) ASC], false, 0
+- Exchange hashpartitioning(coalesce(tmpname#23, ), 200)
+- *Project [name#3 AS tmpname#23, group#4 AS tmpgroup#24, data#5 AS
tmpdata#25]
+- Scan ExistingRDD[name#3,group#4,data#5]
{noformat}
> Unexpected Cartesian product when using eqNullSafe in join with a derived
> table
> -------------------------------------------------------------------------------
>
> Key: SPARK-20073
> URL: https://issues.apache.org/jira/browse/SPARK-20073
> Project: Spark
> Issue Type: Bug
> Components: Optimizer
> Affects Versions: 2.0.2, 2.1.0
> Reporter: Everett Anderson
> Labels: correctness
>
> It appears that if you try to join tables A and B when B is derived from A
> and you use the eqNullSafe / <=> operator for the join condition, Spark
> performs a Cartesian product.
> However, if you perform the join on tables of the same data when they don't
> have a relationship, the expected non-Cartesian product join occurs.
> {noformat}
> // Create some fake data.
> import org.apache.spark.sql.Row
> import org.apache.spark.sql.Dataset
> import org.apache.spark.sql.types._
> import org.apache.spark.sql.functions
> val peopleRowsRDD = sc.parallelize(Seq(
> Row("Fred", 8, 1),
> Row("Fred", 8, 2),
> Row(null, 10, 3),
> Row(null, 10, 4),
> Row("Amy", 12, 5),
> Row("Amy", 12, 6)))
>
> val peopleSchema = StructType(Seq(
> StructField("name", StringType, nullable = true),
> StructField("group", IntegerType, nullable = true),
> StructField("data", IntegerType, nullable = true)))
>
> val people = spark.createDataFrame(peopleRowsRDD, peopleSchema)
> people.createOrReplaceTempView("people")
> scala> people.show
> +----+-----+----+
> |name|group|data|
> +----+-----+----+
> |Fred| 8| 1|
> |Fred| 8| 2|
> |null| 10| 3|
> |null| 10| 4|
> | Amy| 12| 5|
> | Amy| 12| 6|
> +----+-----+----+
> // Now create a derived table from that table. It doesn't matter much what.
> val variantCounts = spark.sql("select name, count(distinct(name, group,
> data)) as variant_count from people group by name having variant_count > 1")
> variantCounts.show
> +----+-------------+
>
> |name|variant_count|
> +----+-------------+
> |Fred| 2|
> |null| 2|
> | Amy| 2|
> +----+-------------+
> // Now try an inner join using the regular equalTo that drops nulls. This
> works fine.
> val innerJoinEqualTo = variantCounts.join(people,
> variantCounts("name").equalTo(people("name")))
> innerJoinEqualTo.show
> +----+-------------+----+-----+----+
>
> |name|variant_count|name|group|data|
> +----+-------------+----+-----+----+
> |Fred| 2|Fred| 8| 1|
> |Fred| 2|Fred| 8| 2|
> | Amy| 2| Amy| 12| 5|
> | Amy| 2| Amy| 12| 6|
> +----+-------------+----+-----+----+
> // Okay now lets switch to the <=> operator
> //
> // If you haven't set spark.sql.crossJoin.enabled=true, you'll get an error
> like
> // "Cartesian joins could be prohibitively expensive and are disabled by
> default. To explicitly enable them, please set spark.sql.crossJoin.enabled =
> true;"
> //
> // if you have enabled them, you'll get the table below.
> //
> // However, we really don't want or expect a Cartesian product!
> val innerJoinSqlNullSafeEqOp = variantCounts.join(people,
> variantCounts("name")<=>(people("name")))
> innerJoinSqlNullSafeEqOp.show
> +----+-------------+----+-----+----+
>
> |name|variant_count|name|group|data|
> +----+-------------+----+-----+----+
> |Fred| 2|Fred| 8| 1|
> |Fred| 2|Fred| 8| 2|
> |Fred| 2|null| 10| 3|
> |Fred| 2|null| 10| 4|
> |Fred| 2| Amy| 12| 5|
> |Fred| 2| Amy| 12| 6|
> |null| 2|Fred| 8| 1|
> |null| 2|Fred| 8| 2|
> |null| 2|null| 10| 3|
> |null| 2|null| 10| 4|
> |null| 2| Amy| 12| 5|
> |null| 2| Amy| 12| 6|
> | Amy| 2|Fred| 8| 1|
> | Amy| 2|Fred| 8| 2|
> | Amy| 2|null| 10| 3|
> | Amy| 2|null| 10| 4|
> | Amy| 2| Amy| 12| 5|
> | Amy| 2| Amy| 12| 6|
> +----+-------------+----+-----+----+
> // Okay, let's try to construct the exact same variantCount table manually
> // so it has no relationship to the original.
> val variantCountRowsRDD = sc.parallelize(Seq(
> Row("Fred", 2),
> Row(null, 2),
> Row("Amy", 2)))
>
> val variantCountSchema = StructType(Seq(
> StructField("name", StringType, nullable = true),
> StructField("variant_count", IntegerType, nullable = true)))
>
> val manualVariantCounts = spark.createDataFrame(variantCountRowsRDD,
> variantCountSchema)
> // Now perform the same join with the null-safe equals operator. This works
> and gives us the expected non-Cartesian product result.
> val manualVarCountsInnerJoinSqlNullSafeEqOp =
> manualVariantCounts.join(people,
> manualVariantCounts("name")<=>(people("name")))
> manualVarCountsInnerJoinSqlNullSafeEqOp.show
> +----+-------------+----+-----+----+
> |name|variant_count|name|group|data|
> +----+-------------+----+-----+----+
> |Fred| 2|Fred| 8| 1|
> |Fred| 2|Fred| 8| 2|
> | Amy| 2| Amy| 12| 5|
> | Amy| 2| Amy| 12| 6|
> |null| 2|null| 10| 3|
> |null| 2|null| 10| 4|
> +----+-------------+----+-----+----+
> {noformat}
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
