[jira] [Created] (SPARK-7564) possible performance bottleneck in SparkSQL's SparkSqlSerializer class

["Noam Barcay (JIRA)" <ji...@apache.org>]

Noam Barcay created SPARK-7564:
----------------------------------

             Summary: possible performance bottleneck in SparkSQL's SparkSqlSerializer class
                 Key: SPARK-7564
                 URL: https://issues.apache.org/jira/browse/SPARK-7564
             Project: Spark
          Issue Type: Bug
          Components: SQL
    Affects Versions: 1.2.1
         Environment: 3 node cluster, each with 100g RAM and 40 cores
            Reporter: Noam Barcay


Running a query over a table that's completely cached in memory I came across surprisingly slow performance. The query is a simple SELECT over a 10Gb table that sits comfortably in memory (Storage tab in Spark UI affirms this). It is held in 60 partitions, to improve parallelism. All operations are over memory, no shuffle is taking place (again, seen via Spark UI).
Looking at some periodic thread-dumps of the workers I see almost all worker threads - 20 in each worker, as the table has 20 partitions in each node - were in one of two states, related to serialization - either trying to acquire an instance of a Kryo serializer from the pool in SparkSqlSerializer, like so:
sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
...
java.util.concurrent.ArrayBlockingQueue.poll(ArrayBlockingQueue.java:361)
com.twitter.chill.ResourcePool.borrow(ResourcePool.java:35)
org.apache.spark.sql.execution.SparkSqlSerializer$.acquireRelease(SparkSqlSerializer.scala:82)
...
org.apache.spark.sql.columnar.InMemoryColumnarTableScan$$anonfun$9$$anonfun$14$$anon$2.next(InMemoryColumnarTableScan.scala:279)

or trying to release one:
sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
...
java.util.concurrent.ArrayBlockingQueue.offer(ArrayBlockingQueue.java:298)
com.twitter.chill.ResourcePool.release(ResourcePool.java:50)
org.apache.spark.sql.execution.SparkSqlSerializer$.acquireRelease(SparkSqlSerializer.scala:86)
...
org.apache.spark.sql.columnar.InMemoryColumnarTableScan$$anonfun$9$$anonfun$14$$anon$2.next(InMemoryColumnarTableScan.scala:279)

I speculate that when contention is high on the underlying ArrayBlockingQueue holding the list of Kryo serialiazers, it starts becoming a bottleneck? Which I would speculate it shouldn't.

The code I'm using (running from Spark shell):
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
hiveContext.sql("SET spark.sql.shuffle.partitions=60").collect()
hiveContext.sql("CACHE TABLE cached_tbl AS SELECT * FROM tbl1 ORDER BY col1").collect() // the ORDER BY here is added to induce a shuffle so the table will be cached in more partitions than it is found in the underlying HDFS storage
hiveContext.sql("select col1, col2, col3 from cached_tbl").saveAsTextFile("/some-path") // the save is actually very quick, it's the select that's taking long





--
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