[SparkSQL]How does spark handle a parquet file in parallel?

[StanZhai <ma...@zhaishidan.cn>]

Hi all,

I'm using Spark (1.4.1) + Hive (0.13.1), I found that a large number of
network IO appeared when query a parquet table *with only one part file* use
SparkSQL. 

The SQL is: SELECT concat(year(fkbb5855f0), "-", month(fkbb5855f0), "-",
day(fkbb5855f0), " 00:00:00"),COUNT(fk919b1d80) FROM test WHERE fkbb5855f0
>= '2015-08-02 00:00:00' AND fkbb5855f0 <
'2015-09-01 00:00:00' AND fk418c5509 IN ('add_summary') AND (fkbb5855f0
!= '' AND fkbb5855f0 is not NULL) GROUP BY year(fkbb5855f0),
month(fkbb5855f0), day(fkbb5855f0)

The SQL's query explain is:

== Parsed Logical Plan ==
'Limit 10000
 'Aggregate ['year('fkbb5855f0),'month('fkbb5855f0),'day('fkbb5855f0)],
['concat('year('fkbb5855f0),-,'month('fkbb5855f0),-,'day('fkbb5855f0),
00:00:00) AS _c0#14,COUNT('fk919b1d80) AS _c1#15]
  'Filter (((('fkbb5855f0 >= 2015-08-02 00:00:00) &&
('fkbb5855f0 < 2015-09-01 00:00:00)) && 'fk418c5509 IN
(add_summary)) && (NOT ('fkbb5855f0 = ) && IS NOT NULL
'fkbb5855f0))
   'UnresolvedRelation [test], None

== Analyzed Logical Plan ==
_c0: string, _c1: bigint
Limit 10000
 Aggregate
[HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFYear(fkbb5855f0#36),HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFMonth(fkbb5855f0#36),HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFDayOfMonth(fkbb5855f0#36)],
[HiveGenericUdf#org.apache.hadoop.hive.ql.udf.generic.GenericUDFConcat(HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFYear(fkbb5855f0#36),-,HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFMonth(fkbb5855f0#36),-,HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFDayOfMonth(fkbb5855f0#36),
00:00:00) AS _c0#14,COUNT(fk919b1d80#34) AS _c1#15L]
  Filter ((((fkbb5855f0#36 >= 2015-08-02 00:00:00) &&
(fkbb5855f0#36 < 2015-09-01 00:00:00)) && fk418c5509#35 IN
(add_summary)) && (NOT (fkbb5855f0#36 = ) && IS NOT NULL
fkbb5855f0#36))
   Subquery test
   
Relation[fkb80bb774#33,fk919b1d80#34,fk418c5509#35,fkbb5855f0#36]
org.apache.spark.sql.parquet.ParquetRelation2@5a271032

== Optimized Logical Plan ==
Limit 10000
 Aggregate
[HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFYear(fkbb5855f0#36),HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFMonth(fkbb5855f0#36),HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFDayOfMonth(fkbb5855f0#36)],
[HiveGenericUdf#org.apache.hadoop.hive.ql.udf.generic.GenericUDFConcat(HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFYear(fkbb5855f0#36),-,HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFMonth(fkbb5855f0#36),-,HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFDayOfMonth(fkbb5855f0#36),
00:00:00) AS _c0#14,COUNT(fk919b1d80#34) AS _c1#15L]
  Project [fkbb5855f0#36,fk919b1d80#34]
   Filter ((((fkbb5855f0#36 >= 2015-08-02 00:00:00) &&
(fkbb5855f0#36 < 2015-09-01 00:00:00)) && fk418c5509#35 INSET
(add_summary)) && (NOT (fkbb5855f0#36 = ) && IS NOT NULL
fkbb5855f0#36))
   
Relation[fkb80bb774#33,fk919b1d80#34,fk418c5509#35,fkbb5855f0#36]
org.apache.spark.sql.parquet.ParquetRelation2@5a271032

== Physical Plan ==
Limit 10000
 Aggregate false, [PartialGroup#42,PartialGroup#43,PartialGroup#44],
[HiveGenericUdf#org.apache.hadoop.hive.ql.udf.generic.GenericUDFConcat(PartialGroup#42,-,PartialGroup#43,-,PartialGroup#44,
00:00:00) AS _c0#14,Coalesce(SUM(PartialCount#41L),0) AS _c1#15L]
  Exchange (HashPartitioning 200)
   Aggregate true,
[HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFYear(fkbb5855f0#36),HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFMonth(fkbb5855f0#36),HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFDayOfMonth(fkbb5855f0#36)],
[HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFYear(fkbb5855f0#36) AS
PartialGroup#42,HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFMonth(fkbb5855f0#36)
AS
PartialGroup#43,HiveSimpleUdf#org.apache.hadoop.hive.ql.udf.UDFDayOfMonth(fkbb5855f0#36)
AS PartialGroup#44,COUNT(fk919b1d80#34) AS PartialCount#41L]
    Project [fkbb5855f0#36,fk919b1d80#34]
     Filter (((((fkbb5855f0#36 >= 2015-08-02 00:00:00)
&& (fkbb5855f0#36 < 2015-09-01 00:00:00)) &&
fk418c5509#35 INSET (add_summary)) && NOT (fkbb5855f0#36 = ))
&& IS NOT NULL fkbb5855f0#36)
      PhysicalRDD
[fkbb5855f0#36,fk919b1d80#34,fk418c5509#35], MapPartitionsRDD[5] at

Code Generation: false
== RDD ==

The size of the `test` table is 3.3GB, I have 5 nodes in the Hadoop cluster,
and Spark use the same cluster. There are 3 replications of test table and
the block size is 64MB. 

The task count of the first stage is 54 when SparkSQL execute the SQL, the
Locality Level of all task is NODE_LOCAL. I use dstat monitoring a node
of the cluster, there are a large number of network IO:

----total-cpu-usage---- -dsk/total- -net/total- ---paging-- ---system--
usr sys idl wai hiq siq| read  writ| recv  send|  in  
out | int   csw
  1   0  99   0   0   0| 107k  389k|
  0     0 | 193B 1097B|2408  5443
  0   0 100   0   0   0|   0     0
|5709B 3285B|   0     0 |1921  4282
  0   0 100   0   0   0|1936k    0 |3761B
1251B|   0     0 |1907  4197
  0   0 100   0   0   0|   0   584k|3399B
1539B|   0     0 |1903  4338
  0   0  99   0   0   0|1936k    0
|4332B 1447B|   0     0 |2070  4448
  4   1  93   3   0   0|  16M  
 0 |5117B 1439B|   0     0 |9177    11k
  1   1  77  21   0   0| 215M    0 |
921k   68M|   0     0 |  21k   10k
  6   1  79  14   0   0| 175M    0 |
327k   19M|   0     0 |  10k 8207
 32   1  68   0   0   0|3308k    0 |
237k   14M|   0     0 |  16k 6536
 22   0  78   0   0   0|2048k    0 |
 98k 5733k|   0     0 |9190  5823
 30   0  69   0   0   0|8080k 8192B| 175k
  11M|   0     0 |  18k 6950
 23   0  77   0   0   0|  18M  
 0 | 727k   52M|   0     0 |  25k 8648
 22   0  78   0   0   0|  28M  
 0 | 920k   96M|   0     0 |  26k   11k
 22   0  78   0   0   0|  31M  
 0 |1003k  114M|   0     0 |  25k   10k
 22   0  78   0   0   0|9372k    0 |
487k   49M|   0     0 |9935  5599
 18   1  81   0   0   0|   0   125k|
 47k 2027k|   0     0 |8820  5358
  4   1  95   0   0   0|  28M  450k|
289k   23M|   0     0 |  16k 7992
  0   0  99   0   0   0|  10M  
 0 | 446k   42M|   0     0 |3765  5262
  0   0 100   0   0   0|1944k    0 |8540B
1364k|   0     0 |1943  4378
  0   0  99   0   0   0|   0   426k|
 11k 2469B|   0     0 |2008  4476
  1   0  99   0   0   0|1852k  368k|
 16k 1687B|   0     0 |2111  4509

But the sql's result size is only 3KB, the network IO confused me. There are
also something confused me too, I found *half of the task's `Input
Size/Records` is `64.0 MB (hadoop) / 0`. * I think Spark dispatch jobs
in a wrong way.

I split the `test` table into different count of part files to valid my
thought. The data is:

part files, single file length，query els，network IO
1, 3.34G, 12s, ~200MB/node
5, 800MB, 10s，~100MB/node
64, 74MB, 8s, ~50MB/node
74, 64MB, 7s, only a little
150, 32MB, 6s, only a little
297，20MB, 5s,  only a little
I also set the replication of `test` table to 5, I found no network IO
appeared, but the query cost 11s! So, I think the network IO should be spark
fetching data from a remote node when a parquet file large than the block
size of Hadoop (64MB).

I want to know:
1. How spark dispatch jobs when a parquet file only have one part file which
large than the block size of Hadoop?
2. In which condition task's `Input Size/Records` is `64.0 MB (hadoop) / 0`
or `5.2 KB (hadoop) / 0` or `0.0 B (hadoop) / 50`?

Any ideas? 
Thanks




--
View this message in context: http://apache-spark-developers-list.1001551.n3.nabble.com/SparkSQL-How-does-spark-handle-a-parquet-file-in-parallel-tp14210.html
Sent from the Apache Spark Developers List mailing list archive at Nabble.com.