Memory and CPU tuning

# tuning_lab.sh: how YARN sizes containers, how many fit, and what happens when the heap is too small
# Run on your PC in the mr-lab folder (it switches the ResourceManager's calculator in part 3).
T=/opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-3.5.0-tests.jar
sleepjob() {  # extra args...
  docker exec -d lab8-namenode-1 bash -c "export HADOOP_CLASSPATH=$T; yarn org.apache.hadoop.mapreduce.SleepJob -libjars $T $* -r 1 -rt 1000 > /tmp/tune.log 2>&1"
}
metrics() { docker exec lab8-namenode-1 curl -s http://resourcemanager:8088/ws/v1/cluster/metrics | python3 -c '
import sys, json; m = json.load(sys.stdin)["clusterMetrics"]
print("   running containers %2d | memory %4d of %d MB | vcores %d of %d" % (m["containersAllocated"], m["allocatedMB"], m["totalMB"], m["allocatedVirtualCores"], m["totalVirtualCores"]))'; }
wait_job() { until docker exec lab8-namenode-1 grep -qE "completed successfully|failed with state" /tmp/tune.log 2>/dev/null; do sleep 3; done; }
bash rm.sh default > /dev/null 2>&1
echo "== 1. a request is rounded UP to a multiple of the minimum allocation (256 MB here)"
for mb in 300 384 600 1100; do
  sleepjob -D mapreduce.map.memory.mb=$mb -m 1 -mt 20000
  sleep 14
  app=$(docker exec lab8-namenode-1 grep -o "application_[0-9_]*" /tmp/tune.log | head -1)
  docker exec lab8-namenode-1 curl -s http://resourcemanager:8088/ws/v1/cluster/apps/$app | python3 -c '
import sys, json; a = json.load(sys.stdin)["app"]
print("map asks %4s MB -> app holds %4d MB = AM 512 + map %d MB" % (sys.argv[1], a["allocatedMB"], a["allocatedMB"] - 512))' $mb
  docker exec lab8-resourcemanager-1 yarn application -kill $app > /dev/null 2>&1; sleep 3
done
echo "== 2. how many maps run at once: 384 MB maps vs 1024 MB maps (24 maps of 30 s each)"
for mb in 384 1024; do
  sleepjob -D mapreduce.map.memory.mb=$mb -m 24 -mt 30000
  sleep 25; echo " maps of $mb MB:"; metrics
  app=$(docker exec lab8-namenode-1 grep -o "application_[0-9_]*" /tmp/tune.log | head -1)
  docker exec lab8-resourcemanager-1 yarn application -kill $app > /dev/null 2>&1; sleep 3
done
echo "== 3. CPU: the default calculator ignores vcores; DominantResourceCalculator counts them"
for calc in default drf; do
  bash rm.sh $calc > /dev/null 2>&1
  sleepjob -D mapreduce.map.memory.mb=384 -m 24 -mt 30000
  sleep 25; echo " calculator $calc:"; metrics
  app=$(docker exec lab8-namenode-1 grep -o "application_[0-9_]*" /tmp/tune.log | head -1)
  docker exec lab8-resourcemanager-1 yarn application -kill $app > /dev/null 2>&1; sleep 3
done
bash rm.sh default > /dev/null 2>&1

# rm.sh (same folder) switches the ResourceManager configuration:
# # rm.sh fifo|capacity|capacity-preempt|capacity-ulf|capacity-ulf-preempt|fair|drf|default : recreate ONLY the ResourceManager with another scheduler
# # (run in the mr-lab folder; HDFS and the NodeManagers keep running)
# mode=$1
# envs="./config"
# case $mode in
#   fifo) envs="$envs, ./sched-fifo.env" ;;
#   capacity) envs="$envs, ./sched-capacity.env" ;;
#   capacity-preempt) envs="$envs, ./sched-capacity.env, ./sched-capacity-preempt.env" ;;
#   capacity-ulf) envs="$envs, ./sched-capacity.env, ./sched-capacity-ulf.env" ;;
#   capacity-ulf-preempt) envs="$envs, ./sched-capacity.env, ./sched-capacity-ulf.env, ./sched-capacity-preempt.env" ;;
#   fair) envs="$envs, ./sched-fair.env" ;;
#   drf) envs="$envs, ./sched-drf.env" ;;
#   default) ;;
# esac
# cat > rm-override.yaml <<Y
# services:
#   resourcemanager:
#     env_file: [$envs]
#     volumes: [./fair-scheduler.xml:/opt/fair-scheduler.xml:ro]
# Y
# docker compose -p lab8 -f docker-compose.yaml -f rm-override.yaml up -d resourcemanager 2>&1 | tail -1
# until [ "$(docker exec lab8-resourcemanager-1 yarn node -list 2>/dev/null | grep -c RUNNING)" = 3 ]; do sleep 3; done
# docker exec lab8-resourcemanager-1 bash -c 'echo "scheduler: $(grep -o "scheduler\.[a-z]*\.[A-Za-z]*Scheduler" /opt/hadoop/etc/hadoop/yarn-site.xml | tail -1)"; mapred queue -list 2>/dev/null | grep -E "Queue Name|Scheduling Info" | paste - - | sed "s/Scheduling Info : //"'

# Run, in the mr-lab folder:
#   bash tuning_lab.sh

# Setting ANY capacity-scheduler key through env makes the image write a capacity-scheduler.xml
# with only those keys, so the queue list must be repeated here (without it the RM fails to start:
# "Queue configuration missing child queue names for root").
CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.queues=default
CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.default.capacity=100
CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.resource-calculator=org.apache.hadoop.yarn.util.resource.DominantResourceCalculator

# Our first version had only the last line. The ResourceManager did not start:
#   docker logs lab8-resourcemanager-1 | grep "missing child queue"

// HeapHog.java: a deliberately memory-hungry map-only job. Each map task keeps EVERY line it reads
// in a list (a common real-world bug: collecting instead of streaming), then writes how many it kept.
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class HeapHog {
  public static class M extends Mapper<LongWritable, Text, NullWritable, Text> {
    private final List<String> kept = new ArrayList<>();
    @Override
    protected void map(LongWritable k, Text line, Context ctx) { kept.add(line.toString()); }
    @Override
    protected void cleanup(Context ctx) throws IOException, InterruptedException {
      Runtime r = Runtime.getRuntime();
      String msg = String.format("kept %,d lines; heap used %d MB of max %d MB", kept.size(),
          (r.totalMemory() - r.freeMemory()) >> 20, r.maxMemory() >> 20);
      System.err.println(msg);
      ctx.write(NullWritable.get(), new Text(msg));
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    String[] rest = new GenericOptionsParser(conf, args).getRemainingArgs();
    Job job = Job.getInstance(conf, "heap hog");
    job.setJarByClass(HeapHog.class);
    job.setMapperClass(M.class);
    job.setNumReduceTasks(0);
    job.setOutputKeyClass(NullWritable.class);
    job.setOutputValueClass(Text.class);
    FileInputFormat.addInputPath(job, new Path(rest[0]));
    FileOutputFormat.setOutputPath(job, new Path(rest[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

// In the mr-lab folder:
//   docker exec lab8-namenode-1 mkdir -p /tmp/mr11
//   docker cp HeapHog.java lab8-namenode-1:/tmp/mr11/
//   docker exec lab8-namenode-1 bash -c "cd /tmp/mr11 && mkdir -p c && javac -cp \$(hadoop classpath) -d c HeapHog.java && jar cf mr11.jar -C c ."
// Then run hog.sh below.

echo "== 4. a map that keeps every line in memory: heap 64 MB vs 300 MB (orders_x30.csv cut into 32 MB splits)"
for heap in 64 300; do
  docker exec lab8-namenode-1 bash -c "cd /tmp/mr11; hdfs dfs -rm -r -f -skipTrash /out/hog$heap > /dev/null 2>&1
    hadoop jar mr11.jar HeapHog -D mapreduce.map.java.opts=-Xmx${heap}m -D mapreduce.input.fileinputformat.split.maxsize=33554432 /user/hadoop/data/orders_x30.csv /out/hog$heap > /tmp/hog.log 2>&1
    echo \" -Xmx${heap}m: \$(grep -oE 'completed successfully|failed with state [A-Z]+' /tmp/hog.log)\"
    grep -oE 'Error: Java heap space|OutOfMemoryError: [A-Za-z ]+|GC overhead limit exceeded' /tmp/hog.log | sort | uniq -c | head -2
    grep -E 'Failed map tasks|Launched map tasks' /tmp/hog.log | sed 's/^\s*/   /'
    hdfs dfs -cat /out/hog$heap/part-m-* 2>/dev/null | sed 's/^/   /'"
done

# Run, in the mr-lab folder:  bash hog.sh