[jira] [Commented] (SPARK-22641) Pyspark UDF relying on column added with withColumn after distinct

2017-11-29 Thread Apache Spark (JIRA) 

[ 
https://issues.apache.org/jira/browse/SPARK-22641?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16271085#comment-16271085
 ] 

Apache Spark commented on SPARK-22641:
--

User 'sethah' has created a pull request for this issue:
https://github.com/apache/spark/pull/19680

> Pyspark UDF relying on column added with withColumn after distinct
> --
>
> Key: SPARK-22641
> URL: https://issues.apache.org/jira/browse/SPARK-22641
> Project: Spark
>  Issue Type: Bug
>  Components: PySpark
>Affects Versions: 2.3.0
>Reporter: Andrew Duffy
>
> We seem to have found an issue with PySpark UDFs interacting with 
> {{withColumn}} when the UDF depends on the column added in {{withColumn}}, 
> but _only_ if {{withColumn}} is performed after a {{distinct()}}.
> Simplest repro in a local PySpark shell:
> {code}
> import pyspark.sql.functions as F
> @F.udf
> def ident(x):
> return x
> spark.createDataFrame([{'a': '1'}]) \
> .distinct() \
> .withColumn('b', F.lit('qq')) \
> .withColumn('fails_here', ident('b')) \
> .collect()
> {code}
> This fails with the following exception:
> {code}
> : org.apache.spark.sql.catalyst.errors.package$TreeNodeException: Binding 
> attribute, tree: pythonUDF0#13
> at 
> org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:56)
> at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:91)
> at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:90)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
> at 
> org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:306)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:304)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:272)
> at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256)
> at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$.bindReference(BoundAttribute.scala:90)
> at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$38.apply(HashAggregateExec.scala:514)
> at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$38.apply(HashAggregateExec.scala:513)
> at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at 
> scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
> at scala.collection.immutable.List.map(List.scala:285)
> at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec.generateResultFunction(HashAggregateExec.scala:513)
> at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec.doProduceWithKeys(HashAggregateExec.scala:659)
> at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec.doProduce(HashAggregateExec.scala:164)
> at 
> org.apache.spark.sql.execution.CodegenSupport$$anonfun$produce$1.apply(WholeStageCodegenExec.scala:85)
> at 
> org.apache.spark.sql.execution.CodegenSupport$$anonfun$produce$1.apply(WholeStageCodegenExec.scala:80)
> at 
> org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:141)
> at 
> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
> at 
> org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:138)
> at 
> org.apache.spark.sql.execution.CodegenSupport$class.produce(WholeStageCodegenExec.scala:80)
> at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec.produce(HashAggregateExec.scala:38)
> at 
> org.apache.spark.sql.execution.WholeStageCodegenExec.doCodeGen(WholeStageCodegenExec.scala:374)
> at 
> org.apache.spark.sql.execution.WholeStageCodegenExec.doExecute(WholeStageCodegenExec.scala:422)
> at 
>

[jira] [Commented] (SPARK-22641) Pyspark UDF relying on column added with withColumn after distinct

2017-11-28 Thread Andrew Duffy (JIRA) 

[ 
https://issues.apache.org/jira/browse/SPARK-22641?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16270132#comment-16270132
 ] 

Andrew Duffy commented on SPARK-22641:
--

Query plan with the literal:

{code}
== Parsed Logical Plan ==
'Project [a#98, b#102, ident('b) AS fails_here#106]
+- Project [a#98, qq AS b#102]
   +- Deduplicate [a#98]
  +- LogicalRDD [a#98], false

== Analyzed Logical Plan ==
a: string, b: string, fails_here: string
Project [a#98, b#102, ident(b#102) AS fails_here#106]
+- Project [a#98, qq AS b#102]
   +- Deduplicate [a#98]
  +- LogicalRDD [a#98], false

== Optimized Logical Plan ==
Aggregate [a#98], [a#98, qq AS b#102, ident(qq) AS fails_here#106]
+- LogicalRDD [a#98], false

== Physical Plan ==
*HashAggregate(keys=[a#98], functions=[], output=[a#98, b#102, fails_here#106])
+- Exchange hashpartitioning(a#98, 200)
   +- BatchEvalPython [ident(qq)], [a#98, pythonUDF0#111]
  +- *HashAggregate(keys=[a#98], functions=[], output=[a#98])
 +- Scan ExistingRDD[a#98]
{code}

And with {{F.col('a')}}

{code}
== Parsed Logical Plan ==
'Project [a#56, b#60, ident('b) AS fails_here#64]
+- Project [a#56, a#56 AS b#60]
   +- Deduplicate [a#56]
  +- LogicalRDD [a#56], false

== Analyzed Logical Plan ==
a: string, b: string, fails_here: string
Project [a#56, b#60, ident(b#60) AS fails_here#64]
+- Project [a#56, a#56 AS b#60]
   +- Deduplicate [a#56]
  +- LogicalRDD [a#56], false

== Optimized Logical Plan ==
Project [a#56, b#60, ident(a#56) AS fails_here#64]
+- Aggregate [a#56], [a#56, a#56 AS b#60, a#56]
   +- LogicalRDD [a#56], false

== Physical Plan ==
*Project [a#56, b#60, pythonUDF0#69 AS fails_here#64]
+- BatchEvalPython [ident(a#56)], [a#56, b#60, pythonUDF0#69]
   +- *Project [a#56, b#60]
  +- *HashAggregate(keys=[a#56], functions=[], output=[a#56, b#60, a#56])
 +- Exchange hashpartitioning(a#56, 200)
+- *HashAggregate(keys=[a#56], functions=[], output=[a#56])
   +- Scan ExistingRDD[a#56]
{code}

> Pyspark UDF relying on column added with withColumn after distinct
> --
>
> Key: SPARK-22641
> URL: https://issues.apache.org/jira/browse/SPARK-22641
> Project: Spark
>  Issue Type: Bug
>  Components: PySpark
>Affects Versions: 2.3.0
>Reporter: Andrew Duffy
>
> We seem to have found an issue with PySpark UDFs interacting with 
> {{withColumn}} when the UDF depends on the column added in {{withColumn}}, 
> but _only_ if {{withColumn}} is performed after a {{distinct()}}.
> Simplest repro in a local PySpark shell:
> {code}
> import pyspark.sql.functions as F
> @F.udf
> def ident(x):
> return x
> spark.createDataFrame([{'a': '1'}]) \
> .distinct() \
> .withColumn('b', F.lit('qq')) \
> .withColumn('fails_here', ident('b')) \
> .collect()
> {code}
> This fails with the following exception:
> {code}
> Py4JJavaError: An error occurred while calling o1321.collectToPython.
> : org.apache.spark.sql.catalyst.errors.package$TreeNodeException: Binding 
> attribute, tree: pythonUDF0#306
>   at 
> org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:56)
>   at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:88)
>   at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:87)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
>   at 
> org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:306)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:304)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:272)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256)
>   at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$.bindReference(BoundAttribute.scala:87)
>   at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$33.apply(HashAggregateExec.scala:475)
>   at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$33.apply(HashAggregateExec.scala:474)
>

[jira] [Commented] (SPARK-22641) Pyspark UDF relying on column added with withColumn after distinct

2017-11-28 Thread Andrew Duffy (JIRA) 

[ 
https://issues.apache.org/jira/browse/SPARK-22641?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16270128#comment-16270128
 ] 

Andrew Duffy commented on SPARK-22641:
--

So it seems this is only a problem when using literal columns. As an example, 
the following riff on the original succeeds:

{code}
import pyspark.sql.functions as F

@F.udf
def ident(x):
return x

spark.createDataFrame([{'a': '1'}]) \
.distinct() \
.withColumn('b', F.col('a')) \
.withColumn('fails_here', ident('b')) \
.collect()
{code}

> Pyspark UDF relying on column added with withColumn after distinct
> --
>
> Key: SPARK-22641
> URL: https://issues.apache.org/jira/browse/SPARK-22641
> Project: Spark
>  Issue Type: Bug
>  Components: PySpark
>Affects Versions: 2.3.0
>Reporter: Andrew Duffy
>
> We seem to have found an issue with PySpark UDFs interacting with 
> {{withColumn}} when the UDF depends on the column added in {{withColumn}}, 
> but _only_ if {{withColumn}} is performed after a {{distinct()}}.
> Simplest repro in a local PySpark shell:
> {code}
> import pyspark.sql.functions as F
> @F.udf
> def ident(x):
> return x
> spark.createDataFrame([{'a': '1'}]) \
> .distinct() \
> .withColumn('b', F.lit('qq')) \
> .withColumn('fails_here', ident('b')) \
> .collect()
> {code}
> This fails with the following exception:
> {code}
> Py4JJavaError: An error occurred while calling o1321.collectToPython.
> : org.apache.spark.sql.catalyst.errors.package$TreeNodeException: Binding 
> attribute, tree: pythonUDF0#306
>   at 
> org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:56)
>   at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:88)
>   at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:87)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
>   at 
> org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:306)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:304)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:272)
>   at 
> org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256)
>   at 
> org.apache.spark.sql.catalyst.expressions.BindReferences$.bindReference(BoundAttribute.scala:87)
>   at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$33.apply(HashAggregateExec.scala:475)
>   at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$33.apply(HashAggregateExec.scala:474)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
>   at 
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
>   at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
>   at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
>   at scala.collection.AbstractTraversable.map(Traversable.scala:104)
>   at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec.generateResultCode(HashAggregateExec.scala:474)
>   at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec.doProduceWithKeys(HashAggregateExec.scala:612)
>   at 
> org.apache.spark.sql.execution.aggregate.HashAggregateExec.doProduce(HashAggregateExec.scala:148)
>   at 
> org.apache.spark.sql.execution.CodegenSupport$$anonfun$produce$1.apply(WholeStageCodegenExec.scala:85)
>   at 
> org.apache.spark.sql.execution.CodegenSupport$$anonfun$produce$1.apply(WholeStageCodegenExec.scala:80)
>   at 
> org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:138)
>   at 
> org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
>   at 
> org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:135)
>   at 
>