TeraSort

TeraSort is a typical Map/Reduce job. It will take map tasks and reduce tasks.

Take the following command to run TeraSort:

hadoop jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-mapreduce-examples.jar terasort \
-Dmapreduce.job.reduces=<REDUCE_TASKS> \
-Ddfs.blocksize=<DFS_BLOCKSIZE>   \
-Dmapreduce.input.fileinputformat.split.minsize=<DFS_BLOCKSIZE> \
-Dio.file.buffer.size=<IO_BUFFER_SIZE> \
-Dmapreduce.map.sort.spill.percent=0.8 \
-Dmapreduce.reduce.shuffle.merge.percent=0.96 \
-Dmapreduce.reduce.shuffle.input.buffer.percent=0.7 \
-Dmapreduce.reduce.input.buffer.percent=0.96 \
/<TERAGEN_DATA_INPUT> \
/<TERASORT_DATA_OUTPUT>

<IO_BUFFER_SIZE> must be equal to your IBM Storage® Scale data pool block size (check this by mmlspool <fs-name> all -L).

<TERAGEN_DATA_INPUT> and <TERASORT_DATA_OUTPUT> must be specified according to your requirements.

<DFS_BLOCKSIZE> could be the default dfs.blocksize from hdfs-site.xml. If you want to take different block size, specify it here. For IBM Storage Scale System or shared storage, <DFS_BLOCKSIZE> cannot be equal to the data pool block size (usually, 1GB block size can give you a good performance). For IBM Storage Scale FPO, the <DFS_BLOCKSIZE> must be equal to your file system data blocksize * blockGroupFactor (check these two values from mmlspool <fs-name> all -L).

<DFS_BLOCKSIZE> impacts the map task number in your cluster. For the above command (mapreduce.input.fileinputformat.split.minsize is specified as <DFS_BLOCKSIZE>), the final map task number is calculated according to <DFS_BLOCKSIZE>. If you have only one file with size 512MB and you specify 500MB as <DFS_BLOCKSIZE>, you will have two splits or map tasks. If you have two 512MB files and you specify 500MB as <DFS_BLOCKSIZE>, you will get four splits or map tasks.

Note: The total split number is the final map task number and cannot be changed by options from TeraSort. If you do not follow the above guide, you might get incorrect split number and therefore impact the map phase in Terasort.

The ideal case for each map task, is that the to-be-processed data size is close but not larger than (70% * mapreduce.map.java.opts * 80%) and this could keep the intermediate data size as small as possible in the shuffle of job.

Very small <DFS_BLOCKSIZE> makes you have more map tasks. Map task number should be proper for the cluster to execute them in one wave or two waves. You should not execute map tasks in three or more waves because this will slow down the performance.

If the map task number is ((MaxTaskPerNode_mem * YarnNodeManagerNumber) - 1), all these map tasks can be handled in one wave. If the map task number is larger than ((MaxTaskPerNode_mem * YarnNodeManagerNumber) – 1), it should be between 1.75 * (MaxTaskPerNode_mem * YarnNodeManagerNumber) and 1.9 * (MaxTaskPerNode_mem * YarnNodeManagerNumber). You could try different map task number by changing file number from <TERAGEN_DATA_INPUT> and <DFS_BLOCKSIZE>.

Usually, <REDUCE_TASKS> should be executed in one wave. That means, <REDUCE_TASKS> should be equal to (TotalReduceTaskPerWave - 1).