Thursday, September 17, 2015

Jenkins + Gitlab + CI

Continuous Integration (CI) is a development practice that requires developers to integrate code into a shared repository several times a day. Each check-in is then verified by an automated build, allowing teams to detect problems early. Jenkins CI is the leading open-source continuous integration server. Hence, we'll deploy Jenkins on our CentOS Server and associate it with the Gitlab repository, so that every time we commit our project via git command, it will be built and deployed automatically by Jenkins.


Install Jenkins On CentOS

Firstly, add Jenkins repository to yum:
$ sudo wget -O /etc/yum.repos.d/jenkins.repo http://jenkins-ci.org/redhat/jenkins.repo
$ sudo rpm --import http://pkg.jenkins-ci.org/redhat/jenkins-ci.org.key

Then, install Jenkins with one-line-command:
$ sudo yum install jenkins

After installation, there're some paths that we need to know:
/usr/lib/jenkins/, the place where jenkins.war resides in;
/etc/sysconfig/jenkins, the configuration file for Jenkins, in which we should modify JENKINS_USER to 'root' in our scenario. Remember to change all the related files and directories to JENKINS_USER correspondingly. And you'll notice that the default port for Jenkins service is 8080, change it if needed.
/var/lib/jenkins, Jenkins home;
/var/log/jenkins/jenkins.log, Jenkins log file.

Finally, we could start Jenkins via command:
$ sudo service jenkins start

The webpage of Jenkins should look like below if everything goes well:


Integrate Jenkins with Gitlab for CI

As for integrating Jenkins with gitlab, there's an official and detailed tutorial [REFERENCE_2] that we could follow. I just wanna take notes on some possible scenarios we may face during the procedure.

Firstly, we need to install GitLab Hook plugin as guided. there's a quick-and-dirty way by install it automatically [Jenkins webpage -> Manage Jenkins -> Manage Plugins], but it might be helpful to install it manually [REFERENCE_3], should any problem occur in that manner. Remember to watch on Jenkins log while restarting Jenkins after installing the plugin we need, in case there are some exceptions complaining.

Then we choose 'Jenkins webpage -> Manage Jenkins -> Configure System' to do the global configuration stuff, where we need to configure Maven, Git and Gitlab module, as depicted below:



Next, we choose 'New Item', and then configure for our project in Jenkins. In the 'Source Code Management' module, we need to set as follows.

The build procedures are threefold, namely, Pre Steps, Build and Post Steps. It is essentially quite self-explained, thus 'talk is cheap and just show the code':



There's a problem, though, concerning executing shell commands in Pre Steps or Post Steps. If the command spawns a subprocess to do its job, it may die after Pre Steps or Post Steps end. So this is a tricky part we need to know about, and the solution is stated in both REFERENCE_4 and REFERENCE_5 with keyword 'BUILD_ID'.

Eventually, we navigate to GitLab services page and activate Jenkins according to REFERENCE_2.


In this time, it will be automatically built and deployed as expected after we commit the changes from local to Gitlab.

But the above manner of configuration has a problem, that when we intend to associate the same Gitlab project to two or more Jenkins services, say, develop environment and production environment, then it fails because only one Jenkins CI could be set in one Gitlab's project.

According to REFERENCE_8, we could notify our Jenkins server via a Restful API provided by Jenkins. Let's configure our project in Jenkins first, or error 'No git jobs using repository:...' may complain [REFERENCE_9]. Go to the Dashboard, click on your project, click Configure, under 'Build Triggers' check the box for 'Poll SCM'. Then select our project in Gitlab, Settings -> Web Hooks, input the Restful API 'http://test.hide.com:8080/git/notifyCommit?url=git@gitlab.com:username/project-name.git' and save. Here, we could add web hooks no matter how many we want.



Security Configuration for Jenkins

Since everyone can access Jenkins if he knows the URL, it needs to be protected via security configuration provided by Jenkins itself. Here's the portal:

REFERENCE_6 is a great tutorial for the above configuring process.



REFERENCE:
1. CentOS 上 Jenkins 安装
2. Gitlab Documentation - Jenkins CI integration
3. How to install a plugin in Jenkins Manually
4. Starting integration tomcat via Jenkins shell script execution
5. Jenkins Doc - ProcessTreeKiller
6. Secure Jenkins with login/password
7. turn Jenkins security off by code
8. Push notification from repository - Git Plugins - Jenkins Doc
9. Jenkins job notification fails with “No git consumers for URI …”


Friday, July 10, 2015

Hive throws Socket Timeout Exception: Read time out

Hive on our gateway applies thrift-server for the purpose of connecting to metadata at remote MySQL database. Yesterday, there was an error complaining about 'MetaException(message:Got exception: org.apache.thrift.transport.TTransportException java.net.SocketTimeoutException: Read timed out)', and it just emerged out of the void.

Firstly, I checked which remote thrift-server current Hive is connecting to via netstat command:
# On terminal_window_1
$ nohup hive -e "use target_db;" &

# On another terminal_window_2
$ netstat -anp | grep 26232

Invoke the second command on terminal_window_2 right after executing the first command, in which, 26232 is the first command's PID. This will end up showing all network connections the specific process is opening:
Proto Recv-Q Send-Q Local Address               Foreign Address                State       PID/Program name
tcp        0      0 127.0.0.1:21786             192.168.7.11:9083              ESTABLISHED 26232/java

Obviously, current hive is connecting to a thrift-server residing at 192.168.7.11:9083. Execute command `telnet 192.168.7.11 9083` to verify whether it is accessible to the target IP and port. In our case, it is accessible.
96:/root>telnet 192.168.7.11 9083
Trying 192.168.7.11...
Connected to undefine.inidc.com.cn (192.168.7.11).
Escape character is '^]'.

Then I switch to that thrift-server, find the process via `ps aux | grep HiveMetaStore`, kill it and then start it again by `hive --service metastore`, it complains 'Access denied for user 'Diablo'@'d82.hide.cn''. Apparently, it is related with MySQL privileges. run the following command in MySQL as user root:
GRANT ALL PRIVILEGES ON Diablo.* TO 'Diablo'@'d82.hide.cn' IDENTIFIED BY 'Diablo'

Restart thrift-server again, and eventually, everything's back to normal.







Wednesday, July 8, 2015

Issue Related With The Owner Of Hive Dirs/Files Created On HDFS is Not The User Executing Hive Command (Hive User Impersonation)

When executing hive command from one of our gateways in any user 'A', then doing some operations which will create files/dirs to hive warehouse on HDFS, the owner of newly-created files/dirs will always be 'supertool', which is the creator of hiveserver2(metastore) process, whichever user 'A' is:
###-- hiveserver2(metastore) belongs to user 'supertool' --
K1201:~>ps aux | grep -v grep | grep metastore.HiveMetaStore --color
500      30320  0.0  0.5 1209800 263548 ?      Sl   Jan28  59:29 /usr/java/jdk1.7.0_11//bin/java -Xmx10000m -Djava.net.preferIPv4Stack=true -Dhadoop.log.dir=/home/workspace/hadoop/logs -Dhadoop.log.file=hadoop.log -Dhadoop.home.dir=/home/workspace/hadoop -Dhadoop.id.str=supertool -Dhadoop.root.logger=INFO,console -Djava.library.path=/home/workspace/hadoop/lib/native -Dhadoop.policy.file=hadoop-policy.xml -Djava.net.preferIPv4Stack=true -Xmx512m -Dhadoop.security.logger=INFO,NullAppender org.apache.hadoop.util.RunJar /home/workspace/hive-0.13.0-bin/lib/hive-service-0.13.0.jar org.apache.hadoop.hive.metastore.HiveMetaStore
K1201:~>cat /etc/passwd | grep 500
supertool:x:500:500:supertool:/home/supertool:/bin/bash

###-- invoke hive command in user 'withdata' and create a database and table --
114:~>whoami
withdata
114:~>hive
hive> create database test_db;
OK
Time taken: 1.295 seconds
hive> use test_db;
OK
Time taken: 0.031 seconds
hive> create table test_tbl(id int);
OK
Time taken: 0.864 seconds

###-- the newly-created database and table belongs to user 'supertool' --
114:~>hadoop fs -ls /user/supertool/hive/warehouse | grep test_db
drwxrwxr-x   - supertool    supertool             0 2015-07-08 15:13 /user/supertool/hive/warehouse/test_db.db
114:~>hadoop fs -ls /user/supertool/hive/warehouse/test_db.db
Found 1 items
drwxrwxr-x   - supertool supertool          0 2015-07-08 15:13 /user/supertool/hive/warehouse/test_db.db/test_tbl

This can be explained by Hive User Impersonation. By default, HiveServer2 performs the query processing as the user who submitted the query. But if related parameters, which are as follows, are set wrongly, the query will run as the user that the hiveserver2 process runs as. The correct way to configure is as below:
<property>
  <name>hive.server2.enable.doAs</name>
  <value>true</value>
  <description>Set this property to enable impersonation in Hive Server 2</description>
</property>
<property>
  <name>hive.metastore.execute.setugi</name>
  <value>true</value>
  <description>Set this property to enable Hive Metastore service impersonation in unsecure mode. In unsecure mode, setting this property to true will cause the metastore to execute DFS operations using the client's reported user and group permissions. Note that this property must be set on both the client and server sides. If the client sets it to true and the server sets it to false, the client setting will be ignored.</description>
</property>

The above settings is self-explained well in their descriptions. Thus there's a need to rectify our hive-site.xml and then restart our hiveserver2(metastore) service.

At this point, there's a puzzling problem that no matter how I change my HIVE_HOME/conf/hive-site.xml, the corresponding setting is not altered at runtime. Eventually, I found that there's another hive-site.xml under HADOOP_HOME/etc/hadoop directory. Consequently, it is advised that we should not put any hive-related configuration files under HADOOP_HOME directory in avoidance of confusion. The official configuration files loading order of precedence can be found at REFERENCE_5.

After revising HIVE_HOME/conf/hive-site.xml, the following commands have guaranteed that the preceding problem is addressed properly.
###-- check runtime hive parameters related with hive user impersonation --
k1227:/home/workspace/hive-0.13.0-bin>hive
hive> set system:user.name;
system:user.name=hadoop
hive> set hive.server2.enable.doAs;
hive.server2.enable.doAs=true
hive> set hive.metastore.execute.setugi;
hive.metastore.execute.setugi=true

###-- start hiveserver2(metastore) again --
k1227:/home/workspace/hive-0.13.0-bin>hive --service metastore
Starting Hive Metastore Server
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.reduce.tasks is deprecated. Instead, use mapreduce.job.reduces
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.min.split.size is deprecated. Instead, use mapreduce.input.fileinputformat.split.minsize
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.reduce.tasks.speculative.execution is deprecated. Instead, use mapreduce.reduce.speculative
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.min.split.size.per.node is deprecated. Instead, use mapreduce.input.fileinputformat.split.minsize.per.node
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.input.dir.recursive is deprecated. Instead, use mapreduce.input.fileinputformat.input.dir.recursive
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.min.split.size.per.rack is deprecated. Instead, use mapreduce.input.fileinputformat.split.minsize.per.rack
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.max.split.size is deprecated. Instead, use mapreduce.input.fileinputformat.split.maxsize
15/07/08 14:28:59 INFO Configuration.deprecation: mapred.committer.job.setup.cleanup.needed is deprecated. Instead, use mapreduce.job.committer.setup.cleanup.needed
15/07/08 14:28:59 WARN conf.HiveConf: DEPRECATED: Configuration property hive.metastore.local no longer has any effect. Make sure to provide a valid value for hive.metastore.uris if you are connecting to a remote metastore.
15/07/08 14:28:59 WARN conf.HiveConf: DEPRECATED: hive.metastore.ds.retry.* no longer has any effect.  Use hive.hmshandler.retry.* instead
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/home/workspace/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/home/workspace/hive-0.13.0-bin/lib/jud_test.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
^Z
[1]+  Stopped                 hive --service metastore
k1227:/home/workspace/hive-0.13.0-bin>bg 1
[1]+ hive --service metastore &
k1227:/home/workspace/hive-0.13.0-bin>ps aux | grep metastore
hadoop    6597 26.6  0.4 1161404 275564 pts/0  Sl   14:28   0:14 /usr/java/jdk1.7.0_11//bin/java -Xmx20000m -Djava.net.preferIPv4Stack=true -Dhadoop.log.dir=/home/workspace/hadoop/logs -Dhadoop.log.file=hadoop.log -Dhadoop.home.dir=/home/workspace/hadoop -Dhadoop.id.str=hadoop -Dhadoop.root.logger=INFO,console -Djava.library.path=/home/workspace/hadoop/lib/native -Dhadoop.policy.file=hadoop-policy.xml -Djava.net.preferIPv4Stack=true -Xmx512m -Dhadoop.security.logger=INFO,NullAppender org.apache.hadoop.util.RunJar /home/workspace/hive-0.13.0-bin/lib/hive-service-0.13.0.jar org.apache.hadoop.hive.metastore.HiveMetaStore
hadoop   11936  0.0  0.0 103248   868 pts/0    S+   14:29   0:00 grep metastore

In which, `set system:user.name` will display current user executing hive command; `set [parameter]` will display the specific parameter's value at runtime. Alternatively, we could list all runtime parameters via `set` command in hive, or from command line: `hive -e "set;" > hive_runtime_parameters.txt`.

A possible exception 'TTransportException: Could not create ServerSocket on address 0.0.0.0/0.0.0.0:9083' will be complained when launching metastore service. According to REFERENCE_6, this is because another metastore or sort of service occupies 9083 port, which is the default port for hive metastore. Kill it beforehand:
k1227:/home/workspace/hive-0.13.0-bin>lsof -i:9083
COMMAND  PID   USER   FD   TYPE     DEVICE SIZE/OFF NODE NAME
java    3499 hadoop  236u  IPv4 3913377019      0t0  TCP *:9083 (LISTEN)
k1227:/home/workspace/hive-0.13.0-bin>kill -9 3499

In this way, we could create database/table again, the owner of corresponding HDFS files/dirs will be changed to the user invoking hive command.



REFERENCE:
1. Setting Up HiveServer2 - Impersonation
2. hive-default.xml.template [hive.metastore.execute.setugi]
3. Hive User Impersonation -mapr
4. Configuring User Impersonation with Hive Authorization - drill
5. AdminManual Configuration - hive [order of precedence]
6. TTransportException: Could not create ServerSocket on address 0.0.0.0/0.0.0.0:9083 - cloudera community




Thursday, July 2, 2015

Encoding Issues Related With Spring MVC Web-Service Project Based On Tomcat

It won't be more agonised and perplexed when we facing with encoding problems in our project. Here's a brief summary on most of (if possible) scenarios encoding problems may occur. Of all the scenarios, I'll take UTF-8 as an instance.

Project Package Encoding

When building our project via maven,  we need to specify charset encoding in the following way:
<build>
  <plugins>
    <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>2.0.2</version>
        <configuration>
            <source>1.7</source>
            <target>1.7</target>
            <encoding>UTF-8</encoding>
        </configuration>
    </plugin>
  </plugins>
</build>

Web Request Encoding

As for web request charset encoding, firstly we need to clarify that there are two kinds of request encoding, namely, 'URI encoding' and 'body encoding'. When we do GET operation upon an URL, all parameters in the URL will be applied 'URI encoding',whose default value is 'ISO-8859-1', whereas POST operation will go through 'body encoding'.

URI Encoding

If we don't intend to change the default charset encoding for URI encoding, we could retrieve the correct String parameter using the following code in SpringMVC controller.
new String(request.getParameter("cdc").getBytes("ISO-8859-1"), "utf-8");

Conversely, the way to change the default charset encoding is to edit '$TOMCAT_HOME/conf/server.xml' file, all <connector> tags in which need to be set `URIEncoding="UTF-8"` as an attribute.
<Connector port="8080" protocol="HTTP/1.1"
           connectionTimeout="20000"
           redirectPort="8443"
           URIEncoding="UTF-8" />
...
<Connector port="8009" protocol="AJP/1.3"
           redirectPort="8443"
           URIEncoding="UTF-8" />
...

Body Encoding

This is set by adding filter in web.xml of your project:
<filter>
    <filter-name>utf8-encoding</filter-name>
    <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
    <init-param>
        <param-name>encoding</param-name>
        <param-value>utf-8</param-value>
    </init-param>
    <init-param>
        <param-name>forceEncoding</param-name>
        <param-value>true</param-value>
    </init-param>
</filter>
<filter-mapping>
    <filter-name>utf8-encoding</filter-name>
    <servlet-name>your_servlet_name</servlet-name>
</filter-mapping>

HttpClient Request Encoding

When applying HttpClient for POST request, there are times that we need to instantiate a StringEntity object. Be sure that you always add encoding information explicitly.
StringEntity se = new StringEntity(xmlRequestData, CharEncoding.UTF_8);

Java String/Byte Conversion Encoding

Similarly to the above StringEntity scenarios in HttpClient Request Encoding, when it comes to talking about String and byte in Java, we always need to specify encoding voluntarily. String has the concept of charset encoding whereas byte does not.
byte[] bytes = "some_unicode_words".getBytes(CharEncoding.UTF_8);
new String(bytes, CharEncoding.UTF_8);

Tomcat catalog.out Encoding

Tomcat's inner logging charset encoding could be set in '$TOMCAT_HOME/bin/catalina.sh'. Find the location of 'JAVA_OPTS' keyword, then append the following setting:
JAVA_OPTS="$JAVA_OPTS -Dfile.encoding=utf-8"

log4j Log File Encoding

Edit 'log4j.properties' file, add the following property under the corresponding appender.
log4j.appender.appender_name.encoding = UTF-8

OS Encoding

Operation System's charset encoding could be checked by `locale` command on Linux. Append the following content to '/etc/profile' will set encoding to UTF-8.
# -- locale --
export LANG=en_US.UTF-8
export LC_CTYPE=en_US.UTF-8
export LC_NUMERIC=en_US.UTF-8
export LC_TIME=en_US.UTF-8
export LC_COLLATE=en_US.UTF-8
export LC_MONETARY=en_US.UTF-8
export LC_MESSAGES=en_US.UTF-8
export LC_PAPER=en_US.UTF-8
export LC_NAME=en_US.UTF-8
export LC_ADDRESS=en_US.UTF-8
export LC_TELEPHONE=en_US.UTF-8
export LC_MEASUREMENT=en_US.UTF-8
export LC_IDENTIFICATION=en_US.UTF-8
export LC_ALL=en_US.UTF-8





REFERENCE:
1. Get Parameter Encoding
2. Spring MVC UTF-8 Encoding

Monday, June 29, 2015

Roaming Through Spark UI And Tune Performance Upon A Specific Use Case

Currently, I'm helping one of my colleagues out a Spark scenario: there are two datasets residing on HDFS. one (alias A) is relatively small in size, with approximately 20,000 line count. Whereas the other one (alias B) is remarkably huge in size, with about 3,000,000,000 lines. The requirement is to calculate the line count of B, whose spid (a kind of id) is in A.

The code is as follows:
val currentDay = args(0)

val conf = new SparkConf().setAppName("Spark-MonitorPlus-LogStatistic")
val sc = new SparkContext(conf)

//---RDD A transforming to RDD[(spid, spid)]---
val spidRdds = sc.textFile("/diablo/task/spid-date/" + currentDay + "-spid-media").map(line =>
line.split(",")(0).trim).map(spid => (spid, spid)).partitionBy(new HashPartitioner(32));
val logRdds: RDD[(LongWritable, Text)] = MzFileUtils.getFileRdds(sc, currentDay, "")
val logMapRdds = MzFileUtils.mapToMzlog(logRdds)

//---RDD B transforming to RDD[(spid, spid)]---
val tongYuanRdd = logMapRdds.filter(kvs => kvs("plt") == "0" && kvs("tp") == "imp").map(kvs => kvs("p").trim).map(spid => (spid, spid)).partitionBy(new HashPartitioner(32));

val filteredTongYuanRdd = tongYuanRdd.join(spidRdds);
println("Total TongYuan Imp: " + filteredTongYuanRdd.count())

It needs to be noticed regarding the usage of both `.partitionBy(new HashPartitioner(32))`. This will guarantee the same key (spid) will be allocated to the same executors when RDD_B joins RDD_A, which is called 'co-partition'. Detailed information on co-partition and co-location is described in REFERENCE_8 (Search keyword 'One is data co-partition, and then data co-location' in that webpage).

Moreover, the above code has a severe bug. If RDD_A has two elements, both of which are (1, 1), whereas RDD_B also has 3 elements of (1, 1). Then after `join` operation, the count will be 2*3=6, which apparently is incorrect when comparing to the correct answer: 3. Thus we need to distinct RDD_A before `join`.
val filteredTongYuanRdd = tongYuanRdd.join(spidRdds.distinct());
println("Total TongYuan Imp: " + filteredTongYuanRdd.count())

When running this code via command `spark-submit --class "com.hide.diablo.sparktools.core.LogAnalysis" --master yarn-cluster --num-executors 32 --driver-memory 8g --executor-memory 4g --executor-cores 4 --queue root.diablo diablo.spark-tools-1.0-SNAPSHOT-jar-with-dependencies.jar 20150515`, there is substantial non-trivial information displayed on Spark UI.

Firstly, navigate to the portal of spark UI for this spark task via applicationId on Yarn monitoring webpage. The following content is all based on Spark 1.2.0.

After entering Spark UI, we are at 'Jobs' tab as default, in which we could see all our jobs generated by Spark according to our transformations on RDD.

When expanding a specific job, it shows all the stages information.

As we can see, there are two maps and one count (include the `join` transformation) stage. Next, expanding into a specific stages, there are two tables exhibiting metrics on executors and tasks respectively.


Apart from 'Jobs' tab, we could also check out all thread dumps for every executor on 'Executors' tab. As a concrete example, the following executor is stuck on `epollWait`, which means it is at the procedure of network IO transmission.


After sketching through most of essential parts of Spark UI, we come back moving on to the above code execution procedure. All stages acts normal until running on 'count' stage, some fails and retries occurs.


Particularly, GC time is relatively high (above 1min) and errors are as below:
org.apache.spark.shuffle.FetchFailedException: Failed to connect toXXXXX/XXXXX:XXXXX
java.lang.OutOfMemoryError: GC overhead limit exceeded
java.lang.OutOfMemoryError: Java heap space

According to REFERENCE_3, we need to enlarge the memory of executors by increasing execution parameter '--executor-memory' to a bigger one, in my case, '14g'. (At this moment, there's an episode that error 'java.lang.IllegalArgumentException: Required executor memory (16384+384 MB) is above the max threshold (15360 MB) of this cluster' may complain after invoking `spark-submit` command, which is solved in REFERENCE_7 and could be referred to a post of mine regarding parameter 'yarn.scheduler.maximum-allocation-mb')

At this time, the spark task is completed in 57min without any OOM complaining.

The above join method is of type reduce-side join, which should be applied on two RDDs whose size are both very large. In this scenario, RDD_A is relatively small, thus a map-side join will increase performance dramatically by diminishing shuffle spilling (For more details on map/reduce-side join, refer to REFERENCE_4 (keyword 'broadcast variables') and REFERENCE_9). Take the reduce-side join case as an example, you could see that shuffle spill (memory) and shuffle spill (disk) is very high:


FYI: Shuffle spill (memory) is the size of the deserialized form of the data in memory at the time when we spill it, whereas shuffle spill (disk) is the size of the serialized form of the data on disk after we spill it. This is why the latter tends to be much smaller than the former. Note that both metrics are aggregated over the entire duration of the task (i.e. within each task you can spill multiple times).

The enhanced solution is to implement via map-side join, which takes advantage of broadcast variable feature in Spark.
val conf = new SparkConf().setAppName("Spark-MonitorPlus-LogStatistic")
val sc = new SparkContext(conf)

//---RDD A transforming to RDD[(spid, spid)]---
val spidRdds = sc.textFile("/diablo/task/spid-date/" + currentDay + "-spid-media").map(line =>
line.split(",")(0).trim).map(spid => (spid, spid)).partitionBy(new HashPartitioner(32));
val logRdds: RDD[(LongWritable, Text)] = MzFileUtils.getFileRdds(sc, currentDay, "")
val logMapRdds = MzFileUtils.mapToMzlog(logRdds)

//---RDD B transforming to RDD[(spid, spid)]---
val tongYuanRdd = logMapRdds.filter(kvs => kvs("plt") == "0" && kvs("tp") == "imp").map(kvs => kvs("p").trim).map(spid => (spid, spid)).partitionBy(new HashPartitioner(32));

val globalSpids = sc.broadcast(spidRdds.collectAsMap());
val filteredTongYuanRdd = tongYuanRdd.mapPartitions({
  iter =>
    val m = globalSpids.value
    for {
      (spid, spid_cp) <- iter
      if m.contains(spid)
    } yield spid
}, preservesPartitioning = true);
println("Total TongYuan Imp: " + filteredTongYuanRdd.count())

In this way, shuffle spilling is drastically mitigated and the execution time reduced to 36min.







REFERENCE:


Friday, June 12, 2015

Troubleshooting On SSH hangs and Machine Stuck When Massive IO-Intensive Processes Are Running

It is inevitable that some users will write and execute devastating code inadvertently on public service-providing machines. Hence, we should enhance the robustness of our machine to the greatest extent.

Today, one of our users forked processes as many as possible, all executing `hadoop get`. As a result, it ate up all of our IO resources even though I've constrained CPU and Memory via `cgroups` upon this specific user. At this time, the phenomenon is that we could neither interact with the prompt nor ssh to this machine anymore. I managed to invoke `top` command before it was fully stuck. This is what it shows:
top - 18:26:10 up 238 days,  5:43,  3 users,  load average: 1782.01, 1824.47, 1680.36
Tasks: 1938 total,   1 running, 1937 sleeping,   0 stopped,   0 zombie
Cpu(s):  2.4%us,  3.0%sy,  0.0%ni,  0.0%id, 94.5%wa,  0.0%hi,  0.0%si,  0.0%st
Mem:  65923016k total, 65698400k used,   224616k free,    13828k buffers
Swap: 33030136k total, 17799704k used, 15230432k free,   157316k cached

As you can see, load average is at an unacceptable peak, and %wa is above 90 (%wa - iowait: Amount of time the CPU has been waiting for I/O to complete). In the meantime, memory has swallowed almost all memory with 17GB swap space being used. Obviously, this is due to the too many processes executing `hadoop get` command.

Eventually, I set the maximum number of processes and open files that the user can fork and open respectively. In this way, conditions will be alleviated to an acceptable scenario.

The configuration is at '/etc/security/limits.conf', we have to edit it in root user. Append the following content to the file:
username      soft    nofile  5000
username      hard    nofile  5000
username      soft    nproc   100
username      hard    nproc   100

Also, there's a nice post regarding how to catch the culprit causing high IO wait in linux.

Reference:
1. Limiting Maximum Number of Processes Available for the Oracle User
2. how to change the maximum number of fork process by user in linux
3. ulimit: difference between hard and soft limits
4. High on %wa from top command, is there any way to constrain it
5. How to ensure ssh via cgroups on centos
6. fork and exec in bash



Saturday, May 30, 2015

Constrain Memory And CPU Via Cgroups On Linux

Limitations on memory and CPU is required inevitably on public-service-provided linux machines like gateways, since there are times that resources on these nodes have been eaten up by clients so that administrators could hardly ssh to the problematic machine when it's stuck.

cgroups (abbreviated from control groups) is a Linux kernel feature that limits, accounts for and isolates the resource usage (CPU, memory, disk I/O, network, etc.) of a collection of processes. And the way to limit a group of users to a specific upperbound for both memory and CPU is as follows.

Firstly, we have to install cgroups just in case it is not on your linux server. In my scenario, the environment is CentOS-6.4 and the following command will just complete the installation process (you need to enable EPEL repository on CentOS of yum command and ensure that CentOS has to be version 6.x).
yum install -y libcgroup

Make it persistent by changing it to 'on' in the runlevel you are using. (Refer to chkconfig and Run Level on Linux)
chkconfig --list | grep cgconfig
chkconfig --level 35 cgconfig on

After that, cgroups is already installed successfully.


Just before we move on to configure limitations via cgroups, we should first put all of our client users in the same group for the purpose of management as a whole. Related commands are as below:
/usr/sbin/groupadd groupname    # add a new group named groupname
usermod -a -G group1,group2 username    # apply username to group1 and group2
groups username    # To check a user's group memberships, use the groups command


Now it's time to configure memory and CPU limitation in '/etc/cgconfig.conf', to which, append the following settings:
group gateway_limit {
    memory {
        memory.limit_in_bytes = 8589934592;
    }

    cpu {
        cpu.cfs_quota_us = 3000000;
        cpu.cfs_period_us = 1000000;
    }
}

Configuration for memory is well self-explained, and that for CPU is defined in this document. In essence, if tasks in a cgroup should be able to access a single CPU for 0.2 seconds out of every 1 second, set cpu.cfs_quota_us to 200000 and cpu.cfs_period_us to 1000000. Note that the quota and period parameters operate on a CPU basis. To allow a process to fully utilize two CPUs, for example, set cpu.cfs_quota_us to 200000 and cpu.cfs_period_us to 100000. If relative share of CPU time is required, setting 'cpu.shares' could be applied. But according to this thread, 'cpu.shares' is work conservingly, i.e., a task would not be stopped from using cpu if there is no competition. If you want to put a hard limit on amount of cpu a task can use, try setting 'cpu.cfs_quota_us' and 'cpu.cfs_period_us'.

After that, we should connect our user/group with the above cgroups limitations in '/etc/cgrules.conf':
@gatewayer      cpu,memory      gateway_limit/

it denotes that for all users in group 'gatewayer', CPU and memory is constrained by 'gateway_limit' which is configured as above.

Notice that if we configure two different lines for the same user/group, then the second line setting will fail, so don't do that:
@gatewayer      memory      gateway_limit/
@gatewayer      cpu      gateway_limit/

Now we have to restart all the related services. Note that we have to switch to path '/etc' in order to execute the following commands without failure, cuz 'cgconfig.conf' and 'cgrules.conf' is available in that path.
service cgconfig restart
service cgred restart  #cgred stands for CGroup Rules Engine Daemon

Eventually, we should verify that all settings have all been hooked up to specific users as expected.

The first way to do that is to login to a user who is in group 'gatewayer', invoking `pidof bash` to check out current PID for the bash terminal. Then `cat /cgroup/cpu/gateway_limit/cgroup.procs` as well as `cat /cgroup/memory/gateway_limit/cgroup.procs` to see whether the former PID is in here. If so, it means memory and CPU is monitored by cgroups for the current user.

The second way to achieve this is a little bit simpler, for you only have to login to a user, execute `cat /proc/self/cgroup`. If gateway_limit is applied both to memory and CPU as follows, then it is well configured.
246:blkio:/
245:net_cls:/
244:freezer:/
243:devices:/
242:memory:/gateway_limit
241:cpuacct:/
240:cpu:/gateway_limit
239:cpuset:/

As for testing, here's a memory-intensive python script that could be used to test memory limitation:
import string
import random
import time

if __name__ == '__main__':
    d = {}
    i = 0;
    for i in range(0, 900000000):
        d[i] = some_str = ' ' * 512000000
        if i % 10000 == 0:
            print i

When monitoring via `ps aux | grep this_script_name`, the process is killed when the consuming cpu exceeds the upperbound set in cgroups. FYI, swappiness is currently set to 0, thus no swap space will be used when physical memory is exhausted and the process will be killed by Linux kernel. If we intend to make our Linux environment more elastic, we could modify swappiness to a higher level (default is 60).

For CPU-intensive test, referencing to this thread, the following command will be executed on the specific user to create multiple processes consuming CPU time. On another terminal window, we could execute `top -b -n 1 | grep current_user_name_or_dd | awk '{s+=$9} END {print s}'` to sum up all the dd processes CPU time.
fulload() { dd if=/dev/zero of=/dev/null | dd if=/dev/zero of=/dev/null | dd if=/dev/zero of=/dev/null | dd if=/dev/zero of=/dev/null & }; fulload; read; killall dd



Reference:
1. cgroups documentation
2. cgroups on centos6
3.1. cgrules.conf(5) - Linux man page
3.2. cgconfig.conf(5) - Linux man page
4. how-to-create-a-user-with-limited-ram-usage - stackexchange
5. how-can-i-configure-cgroups-to-fairly-share-resources-between-users - stackexchange
6. how-can-i-produce-high-cpu-load-on-a-linux-server - superuser
7. shell-command-to-sum-integers-one-per-line - stackoverflow
8. Why does Linux swap out pages when I have many pages cached and vm.swappiness is set to 0 - quora