Spark and Parquet with large block size

One of issue when I run a Spark application in yarn cluster mode is that my executor container is killed because the memory exceeds memory limits. NodeManager's log shows the folowing messages:

2014-12-22 09:22:48,906 INFO org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl: Memory usage of ProcessTree 26453 for container-id container_1418853356063_0165_01_000005_01: 45.8 GB of 44.5 GB physical memory used; 46.5 GB of 93.4 GB virtual memory used
2014-12-22 09:22:48,907 WARN org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl: Process tree for container: container_1418853356063_0165_01_000005_01 has processes older than 1 iteration running over the configured limit. Limit=47781511168, current usage = 49199300608
2014-12-22 09:22:48,907 WARN org.apache.hadoop.yarn.server.nodemanager.containermanager.monitor.ContainersMonitorImpl: Container [pid=26453,containerID=container_1418853356063_0165_01_000005_01] is running beyond physical memory limits. Current usage: 45.8 GB of 44.5 GB physical memory used; 46.5 GB of 93.4 GB virtual memory used. Killing container.
Dump of the process-tree for container_1418853356063_0165_01_000005_01 :
        |- PID PPID PGRPID SESSID CMD_NAME USER_MODE_TIME(MILLIS) SYSTEM_TIME(MILLIS) VMEM_USAGE(BYTES) RSSMEM_USAGE(PAGES) FULL_CMD_LINE
        |- 26526 26453 26453 26453 (java) 315870 8116 49777754112 12011233 /usr/java/jdk1.7.0_67-cloudera/bin/java -server -XX:OnOutOfMemoryError=kill %p -Xms45056m -Xmx45056m -Djava.io.tmpdir=/grid/2/yarn/nm/usercache/bwang/appcache/application_1418853356063_0165/container_1418853356063_0165_01_000005_01/tmp org.apache.spark.executor.CoarseGrainedExecutorBackend akka.tcp://sparkDriver@hadoop-data:48923/user/CoarseGrainedScheduler 3 hadoop-data 23
        |- 26453 26450 26453 26453 (bash) 1 1 108658688 315 /bin/bash -c /usr/java/jdk1.7.0_67-cloudera/bin/java -server -XX:OnOutOfMemoryError='kill %p' -Xms45056m -Xmx45056m  -Djava.io.tmpdir=/grid/2/yarn/nm/usercache/bwang/appcache/application_1418853356063_0165/container_1418853356063_0165_01_000005_01/tmp org.apache.spark.executor.CoarseGrainedExecutorBackend akka.tcp://sparkDriver@hadoop-data:48923/user/CoarseGrainedScheduler 3 hadoop-data 23 1> /var/log/hadoop-yarn/container/application_1418853356063_0165/container_1418853356063_0165_01_000005_01/stdout 2> /var/log/hadoop-yarn/container/application_1418853356063_0165/container_1418853356063_0165_01_000005_01/stderr

If an executor is lost, Spark will spawn another executor to recover from this failure. Under this situation, any persisted or cached RDDs are lost too, Spark must re-process from the beginning. This usually leads to a lengthy processing time. In any case, you should avoid this situation happening. For example, one of my application can finish in 5 mins if no lost executor happens, otherwise it could take more than 40 mins.

Because Spark builds an in-memory hash when doing groupByKey or combineByKey, which needs a lot of memory. One of suggestion is to use a larger parallelism to make the reduce task smaller.

If your application read Parquet files, you must be careful to choose executor memory and cores by taking into account of the block size of parquet files because reading Parquet files may consume a lot of memory. For example, Impala writes Parquet files using 1GB HDFS blocksize by default. My application needs to read 136 parquet files output by Impala. The application runs on 4 nodes with 24 virtual cores each node using "--executor-memory 44G --executor-cores 23 --num-executors 4", and my executors will be killed. But if I use 10 cores per executor "--executor-cores 10", everything passes through without any executor is lost. The reason is when you read 136 parquet files, there are 23 running tasks in one executor at the same time, so 23GB are allocated for reading Parquet, and 23 tasks try to build hash in memory. By using 10 cores, there are more memory for one task, and each task has enough memory to build the hash. Even the job is a little bit slow, it actually save time because no re-computation due to the lost executors.

"NoSuchMethodError: com.google.common.collect.Sets.newConcurrentHashSet()Ljava/util/Set;" when running spark-cassandra-connector

When I tried to update cassandra table using spark-cassandra-connector in a Spark application, I encountered this problem. The reason is that there are multiple versions of com.google.guava:guava exists. I'm using CDH 5.1.0 with Spark 1.0.0. Spark uses guava-14.0.1, Hadoop mapreduce use guava-11.0.2, and spark-cassandra-connector uses guava-15.0.0. The similar issue is reported here: https://github.com/datastax/spark-cassandra-connector/issues/292 and https://github.com/datastax/spark-cassandra-connector/issues/326 I tried to use spark.files.userClassPathFirst=true, there are other errors. I tried to put guava-15.0 jar to SPARK_CLASSPATH, the driver side didn't report error, but failed on the worker side. Actually the solution is very simple, in you Spark project, exclude guava from spark-cassandra-connector.

  
   com.datastax.spark
   spark-cassandra-connector_${scala.major}
   
    
     com.codahale.metrics
     metrics-core
    
    
     com.google.guava
     guava
    
   
   1.1.0-beta1
  

When you run spark-submit, don't put guava-15.0 to --jars or the classpath.

Build scalatest-eclipse-plugin for Scala-IDE 4.0.0 RC1

I upgraded Scala-IDE in Eclipse Luna to 4.0.0 RC1, but didn't find Scalatest plugin. It is inconvenient and time consuming to run "mvn test" every time I change the test. I decided to build Scalatest plugin by myself. It turns out pretty easy, here is how:

• Clone from github: git clone https://github.com/scalatest/scalatest-eclipse-plugin.git
• Use branch kepler-nightly-2.11: git checkout -b nightly-2.11 kepler-nightly-2.11
• Change luna profile in scalatest-eclipse-plugin/pom.xml: replace "ecosystem/next" with "sdk".

      
        scala-ide-4.0-2_11-luna
        
          2.11.2
          2_11
          2.11
          v-4
          ${repo.scala-ide.root}/sdk/lithium/e44/scala211/dev/site/
        
      
  

• Build: mvn clean package
• Install in eclipse, use "file:/${path}/scalatest-eclipse-plugin/org.scala-ide.sdt.scalatest.update-site/target/site/" in "Install new software ..."

Solarized Light (Scala) for Eclipse

Just install ScalaIDE 4.0.0-RC1 on my Eclipse Luna. I really like "Solarized Light (Scala)" color theme as shown in the release note.

• You need to install "Eclipse Color Theme Plugin", otherwise you could not find "Prefrences->General->Appearance->Color Theme".
• You need to download the theme from this page. Just click Eclipse Color Theme (XML) - for Eclipse Color Theme Plugin.
• Go to "Preferences->General->Appearance->Color Theme, click the button "Import a theme", choose the download file and import it.
• Then you may find that not all syntax color are same as the picture in the release note. For example, class and trait names are not colored. If so, you need to check the configuration in "Prefrences->Scala->Editor->Syntax Coloring". Not all "Semantic" item are enabled by default. Just enable them and everything will be colored.

Add Terminals View

I really like Eclipse Luna's new feature "Terminals". It is convenient so that you don't have to leave Eclipse to open a terminal to type your command. And Mylyn task still record the time you spend on command lines. Each time when you click "Window->Show View->Other->Terminals", a new terminal view is created, and you can open one or more terminal in that view. I'd like open ONLY one terminal and run gnu screen in it so that I can switch to a new screen or scroll up and down without moving the mouse. There is another benefit when doing that, even you restart Eclipse, you won't lost your screen session. Just open a new Terminals view and resume the session using "screen -R". But there is one issue: a new blank terminals view is created when you are in a new perspective and try to "show view", the terminals you already open won't show up. How can I share the screen session among different perspectives, for example, java/java ee/scala? Here is how you can fix it: don't stop there, just click "Open a terminal" button in the blank terminals view and click OK to create a new local terminal. In my eclipse, a view "Terminals 1" is shown up with two terminals, one is running gnu screen and the other is new created. Once you have the view "Terminals 1", you can close the blank terminals view. This sounds a bug to me.

Impala-shell may have control sequence in its output

Assume you have a table has three rows, what is the result in the output file /tmp/my_table_count? 3? Actually it is not. There is a control sequence "ESC[?1034h" on my terminal.

$ impala-shell -B -q "select count(1) from my_table" > /tmp/my_table_count
$ xxd /tmp/my_table_count
0000000: 1b5b 3f31 3033 3468 320a                 .[?1034h2.

It will cause a problem when you use the result in a script which tries to update a partition's numRows in Impala.

  local a=$(impala-shell -B -q "select count(1) from my_table where part_col='2014-08-28'")
  impala-shell -q "alter table my_table partition(part_col='2014-08-20') set tblproperties('numRows'='$a')"

If you run this script, you will get the wrong value for #Rows due to the escape control sequence.

Query: show table stats my_table
+------------+-------+--------+--------+--------------+---------+
| part_col   | #Rows | #Files | Size   | Bytes Cached | Format  |
+------------+-------+--------+--------+--------------+---------+
| 2014-08-28 | -1    | 2      | 2.87KB | NOT CACHED   | PARQUET |
| Total      | -1    | 2      | 2.87KB | 0B           |         |
+------------+-------+--------+--------+--------------+---------+
Returned 2 row(s) in 0.06s

You can fix it by unsetting TERM like this:

  local a=$(TERM= impala-shell -B -q "select count(1) from my_table where part_col='2014-08-28'")
  impala-shell -q "alter table my_table partition(part_col='2014-08-20') set tblproperties('numRows'='$a')"

Set replication for files in Hadoop

• Change the existing files's replications

  hadoop fs -setrep -R -w 2 /data-dir
  

• Set replication when loading a file

  hadoop fs -Ddfs.replication=2 -put local_file dfs_file