[jira] [Created] (SPARK-11368) Spark scan all partitions when using Python UDF and filter over partitioned column is given

["Maciej Bryński (JIRA)" <ji...@apache.org>]

Maciej Bryński created SPARK-11368:
--------------------------------------

             Summary: Spark scan all partitions when using Python UDF and filter over partitioned column is given
                 Key: SPARK-11368
                 URL: https://issues.apache.org/jira/browse/SPARK-11368
             Project: Spark
          Issue Type: Bug
          Components: PySpark, SQL
            Reporter: Maciej Bryński
            Priority: Critical


Hi,
I think this is huge performance bug.

I created parquet file partitioned by column.
Then I make query with filter over partition column and filter with UDF.
Result is that all partition is scanned.

Sample data:
{code}
rdd = sc.parallelize(range(0,10000000)).map(lambda x: Row(id=x%1000,value=x)).repartition(1)
df = sqlCtx.createDataFrame(rdd)
df.write.mode("overwrite").partitionBy('id').parquet('/mnt/mfs/udf_test')
df = sqlCtx.read.parquet('/mnt/mfs/udf_test')
df.registerTempTable('df')
{code}
Then queries:
Without udf - Spark reads only 10 partitions:
{code}
start = time.time()
sqlCtx.sql('select * from df where id >= 990 and value > 100000').count()
print(time.time() - start)

0.9993703365325928
{code}
With udf Spark reads all the partitions:
{code}
sqlCtx.registerFunction('multiply2', lambda x: x *2 )
start = time.time()
sqlCtx.sql('select * from df where id >= 990 and multiply2(value) > 200000').count()
print(time.time() - start)

13.0826096534729
{code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org