# TeraSort benchmark 

MapReduce tera tests ensure that MR is running correctly.

Depending on the parameters and the numbers of time it is ran in a row, the test
suite can be used to assess regularity, balance or workload on a cluster.

    module.exports = header: 'Benchmark', handler: ->
      {benchmark} = @config.ryba
      node_output_header = "Number of maps,Number of rows,Bytes written, Bytes read, Blocks written, Blocks read, Blocks replicated, Blocks removed, Blocks replicated"
      username = /^(.*?)(@|$)/.exec(benchmark.kerberos.principal)[1]
      teragen_output_dir = "/user/#{username}/benchmark_teragen"
      teragen_output_file = "#{benchmark.output}/teragen.csv"
      terasort_output_dir = "/user/#{username}/benchmark_terasort"
      terasort_output_file = "#{benchmark.output}/terasort.csv"
      job_output_header = "Number of maps,Number of rows,#{benchmark.terasort.stdout_value_names.join ","},Real time"

## Prepare job output files

      @call header: 'Preparation', handler: ->
        @file
          ssh: false
          target: teragen_output_file
          content: job_output_header
          unless_exists: true
        @file
          ssh: false
          target: terasort_output_file
          content: job_output_header
          unless_exists: true
        @file (
          ssh: false
          target: "#{benchmark.output}/#{node.name}.csv"
          content: node_output_header
          unless_exists: true
        ) for node in benchmark.datanodes

Run each jobs N times (defined by parameter `iterations`)

      @each benchmark.terasort.parameters, (options) ->
        parameters = options.key
        @each [0...benchmark.iterations], (options) ->
          iteration = options.key

## Clean HDFS directories 

          @system.execute
            header: "#{iteration} HDFS Cleanup"
            cmd: """
            echo #{benchmark.kerberos.password} | kinit #{benchmark.kerberos.principal} >/dev/null 2>&1
            hdfs dfs -rm -r -skipTrash '#{teragen_output_dir}' '#{terasort_output_dir}'
            """
            code_skipped: 1
            trap: true

## DN Metrics Before 

Before (and after) each test, request the following information for every 
datanode using the node's JMX interface :

* Bytes written 
* Bytes read 
* Blocks written 
* Blocks read 
* Blocks replicated 
* Blocks removed  
* Blocks validated 

These can be used to validate data repartition in the cluster.

          @call header: "#{iteration} Collect Before", handler: ->
            @each benchmark.datanodes, (options, cb) ->
              node = options.key
              @system.execute
                cmd: """
                echo #{benchmark.kerberos.password} | kinit #{benchmark.kerberos.principal} >/dev/null 2>&1
                curl --fail -k --negotiate -u: \
                  -H "Content-Type: application/json" \
                  -X GET #{node.urls.metrics}
                """
              , (err, execute, stdout) ->
                throw err if err
                data = JSON.parse stdout
                throw Error "Invalid Response" unless new RegExp("Hadoop:service=DataNode,name=DataNodeActivity-#{node.name}-1004").test data?.beans[0]?.name
                @file
                  ssh: false
                  target: "#{benchmark.output}/#{node.name}.csv"
                  content: "\n#{parameters.maps},#{parameters.rows},#{parse_datanode_jmx data.beans[0]}"
                  append: true
              @next cb

## TeraGen 

          @call header: "#{iteration} Teragen", handler: ->
            @system.execute
              cmd: """
              echo #{benchmark.kerberos.password} | kinit #{benchmark.kerberos.principal} >/dev/null 2>&1
              time hadoop jar #{benchmark.jars.current} \
                teragen \
                -Dmapreduce.job.maps=#{parameters.maps} \
                '#{parameters.rows}' \
                '#{teragen_output_dir}'
              """
            , (err, _, stdout, stderr) ->
              @file
                unless: !!err
                ssh: false
                target: teragen_output_file
                content: parse_metrics stderr, parameters.maps, parameters.rows
                append: true

## TeraSort

          @call header: "#{iteration} Terasort", handler: ->
            @system.execute
              cmd: """
              echo #{benchmark.kerberos.password} | kinit #{benchmark.kerberos.principal} >/dev/null 2>&1
              time hadoop jar #{benchmark.jars.current} \
                terasort \
                -Dmapreduce.job.maps=#{parameters.maps} \
                '#{teragen_output_dir}' \
                '#{terasort_output_dir}'
              """
            , (err, _, stdout, stderr) ->
              @file
                unless: !!err
                ssh: false
                target: terasort_output_file
                content: parse_metrics stderr, parameters.maps, parameters.rows
                append: true

## DN Metrics After

          @call header: "#{iteration} Collect After", handler: ->
            @each benchmark.datanodes, (options, cb) ->
              node = options.key
              @system.execute
                cmd: """
                curl  --fail -k --negotiate -u: \
                  -H "Content-Type: application/json" -k \
                  -X GET #{node.urls.metrics}
                """
              , (err, execute, stdout) ->
                throw err if err
                data = JSON.parse stdout
                throw Error "Invalid Response" unless new RegExp("Hadoop:service=DataNode,name=DataNodeActivity-#{node.name}-1004").test data?.beans[0]?.name
                @file
                  ssh: false
                  target: "#{benchmark.output}/#{node.name}.csv"
                  content: "\n#{parameters.maps},#{parameters.rows},#{parse_datanode_jmx data.beans[0]}"
                  append: true
              @next cb

## Utils 

Parse the output of the Datanode JMX query run before and after each teragen /
terasort job

      parse_datanode_jmx = (bean) ->
        line = ""
        line += "#{bean.BytesWritten},"
        line += "#{bean.BytesRead},"
        line += "#{bean.BlocksWritten},"
        line += "#{bean.BlocksRead},"
        line += "#{bean.BlocksReplicated},"
        line += "#{bean.BlocksRemoved},"
        line += "#{bean.BlocksVerified}"
        line

Parse the output of a teragen / terasort job to retrieve job metrics

      parse_metrics = (output, maps, rows) ->
        metrics = [
          maps
          rows
        ]
        for line in output.split "\n"
          if /.*Running job: .*/.test line
            metrics.applicationId = line.replace /.*Running job: /, ""
          for name in benchmark.terasort.stdout_value_names
            unless line.indexOf(name) == -1
              metrics.push line.split("=")[1]
          unless line.indexOf("real") == -1 # job duration
            metrics.push line.substring line.lastIndexOf "\t"
        "\n#{metrics.join ','}"