Hadoop Streaming示例程序(wordcount)

run_hadoop_word_counter.sh

$HADOOP_BIN streaming \-input "${INPUT}" \-output "${OUT_DIR}" \-cacheArchive "${TOOL_DIR}/python2.7.2.tgz""#." \-file "mapper_word_counter.py" \                                                                                                                                         -file "reducer_word_counter.py" \-file "filter_word_counter.py" \-mapper "./python2.7.2/bin/python mapper_word_counter.py" \-combiner "./python2.7.2/bin/python reducer_word_counter.py" \-reducer "./python2.7.2/bin/python reducer_word_counter.py" \-jobconf abaci.job.base.environment="centos6u3_hadoop" \-jobconf mapred.job.priority="NORMAL" \-jobconf mapred.job.name="${TASK_NAME}" \-jobconf mapred.map.tasks="${MAP_NUM}" \-jobconf mapred.reduce.tasks="${REDUCE_NUM}" \-jobconf mapred.map.memory.limit="1000" \-jobconf mapred.reduce.memory.limit="1000" \-jobconf mapred.job.map.capacity="3000" \-jobconf mapred.job.reduce.capacity="2500" \-jobconf mapred.job.keep.files.hours=12 \-jobconf mapred.max.map.failures.percent=1 \-jobconf mapred.reduce.tasks.speculative.execution="false"

mapper_word_counter.py

import sys for line in sys.stdin:fields = line.strip().split('\t')try:cnt = 1                                                                                                                                                              dateval = fields[1]sys.stdout.write('%s\t%d\n' %(dateval, cnt))except Exception as exp:sys.stderr.write("exp:%s, %s" %(str(exp), line))

reducer_word_counter.py

import sys word_pre = None
counter_pre = 0 for line in sys.stdin:try:word, cnt  = line.strip().split('\t')                                                                                                                                cnt = int(cnt)except Exception as exp:sys.stderr.write('Exp:%s,line:%s' %(str(exp), line.strip()))continueif word == word_pre:counter_pre += cnt else:if word_pre:print('%s\t%d' %(word_pre, counter_pre))word_pre = wordcounter_pre = cnt if word_pre:print('%s\t%d' %(word_pre, counter_pre))

纯文本输入格式

• 每个mapper输入若干行
  -inputformat "org.apache.hadoop.mapred.TextInputFormat"
• 指定每个mapper输入的行数
  -inputformat "org.apache.hadoop.mapred.lib.NLineInputFormat" -jobconf mapred.line.input.format.linespermap="5"

文件分发方式：

-file将客户端本地文件打成jar包上传到HDFS然后分发到计算节点；
-cacheFile将HDFS文件分发到计算节点；
-cacheArchive将HDFS压缩文件分发到计算节点并解压；

分桶&排序

Hadoop默认会把map输出行中遇到的第一个分隔符(默认为\t)前面的部分作为key，后面的作为value，如果输出行中没有指定的分隔符，则整行作为key，value被设置为空字符串。mapper输出的key经过partition分发到不同的reduce里。

• 应用示例

${HADOOP_BIN} streaming \-input "${INPUT}" \-output "${OUT_DIR}" \-mapper cat \-reducer cat \-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \-jobconf mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \-jobconf stream.num.map.output.key.fields=4 \-jobconf stream.map.output.field.separator=. \-jobconf map.output.key.field.separator=. \-jobconf mapred.text.key.partitioner.options=-k1,2 \-jobconf mapred.text.key.comparator.options="-k3,3 -k4nr" \-jobconf stream.reduce.output.field.separator=. \-jobconf stream.num.reduce.output.key.fields=4 \-jobconf mapred.reduce.tasks=5

说明：

• 设定mapper输出的key
  stream.map.output.field.separator 设置map输出的字段分隔符
  stream.num.map.output.key.fields 设置map输出的前几个字段作为key
• 设定根据key进行分桶的规则
  org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner partition类
  map.output.key.field.separator 设置key内的字段分隔符(KeyFieldBasedPartitioner和KeyFieldBasedComparator所特有)
  num.key.fields.for.partition 设置key内前几个字段用来做partition
  mapred.text.key.partitioner.options 可单独指定key中哪些字段做partition，和num.key.fields.for.partition一起使用以num.key.fields.for.partition为准
• 设定根据key进行排序的规则
  KeyFieldBasedComparator 可灵活设置的高级比较器，默认使用Text的基于字典序或者通过-n来基于数字比较
  mapred.text.key.comparator.options 设置key中需要比较的字段或字节范围
• 设定reducer输出的key
  stream.reduce.output.field.separator 设置reduce输出的字段分隔符
  stream.num.reduce.output.key.fields 设置reduce输出的前几个字段作为key

多路输出

Hadoop支持多路输出，可以将MapReduce的处理数据输出到多个part-xxxxx-X文件中(X是A-Z共26个字母中的一个)。程序需要在maper(正对仅有mapper的MR任务)/reducer(针对包含reducer的任务)程序中将输出形式由<key,value>变为<key, value#X>，以便输出特定后缀的文件中。其中#X仅仅用做指定输出文件后缀， 不会出现在输出内容中。
启动脚本中需要指定
-outputformat org.apache.hadoop.mapred.lib.SuffixMultipleTextOutputFormat
或者
-outputformat org.apache.hadoop.mapred.lib.SuffixMultipleSequenceFileOutputFormat

• 应用示例
  run_hadoop.sh

${HADOOP_BIN} streaming \-input "${INPUT}" \-output "${OUT_DIR}" \-cacheArchive "${TOOL_DIR}/python2.7.2.tgz""#." \-file "mapper_worker.sh" \-file "reducer_worker.py" \-mapper "sh mapper_worker.sh" \-reducer "python2.7.2/bin/python reducer_worker.py" \-inputformat "org.apache.hadoop.mapred.TextInputFormat" \-outputformat "org.apache.hadoop.mapred.lib.SuffixMultipleTextOutputFormat" \-jobconf mapred.job.priority="NORMAL" \-jobconf mapred.job.name="${TASK_NAME}" \-jobconf mapred.map.tasks="${MAP_NUM}" \-jobconf mapred.reduce.tasks="${REDUCE_NUM}" \-jobconf mapred.max.split.size=134217728 \-jobconf mapred.map.memory.limit="800" \-jobconf mapred.reduce.memory.limit="500" \-jobconf mapred.job.map.capacity="3500" \-jobconf mapred.job.reduce.capacity="2000" \-jobconf mapred.job.keep.files.hours=12 \-jobconf mapred.max.map.failures.percent=1 \-jobconf mapred.reduce.tasks.speculative.execution="false"

reducer_worder.py

for line in sys.stdin:record = line.strip()fields = record.split('\t')if len(fields) != 7:continuevcpurl, playurl, title, poster, duration, pubtime, accept = fieldsduration = int(duration)pubtime = int(pubtime)accept = int(accept)if duration < 60:sys.stdout.write('%s#A\n' %(record))elif duration < 300：sys.stdout.write('%s#B\n' %(record))else:sys.stdout.write('%s#C\n' %(record))

本地调试

为避免在启动MR任务后才发现程序bug，最好提前在本地模拟MR的运行流程，验证结果是否符合预期

cat inputfile | ./mapper_task.sh | sort -t$'\t' -k1,1 | ./reducer.sh

压缩输出

Hadoop默认支持gzip压缩, streaming作业中指定以下参数即可使输出以gzip形式压缩.

-D mapreduce.output.fileoutputformat.compress=true
-D mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.GzipCodec

Hadoop 是可自行读取gzip压缩的数据，无需特殊指明输入是 Gzip 压缩。Gzip 的特点是压缩比较高，Hadoop 原生支持，缺点是压缩效率并不是很高，压缩比和效率不可兼得，需要考虑其他压缩方式。

Hadoop常用配置项

Hadoop环境下系统变量

• 变量名列表

https://hadoop.apache.org/docs/r1.2.1/mapred_tutorial.html#Configured+Parameters

• 应用示例：
  Shell版

#!/bin/bashset -o pipefail
HOST="localhost"
PORT=$((1000 + ${mapred_task_partition}))awk '{print $2}' \| ./access_remote_data ${HOST} ${PORT} outdata.gzhdfs_outfile=${mapred_work_output_dir}/${mapred_task_partition}.pack
cat outdata.gz \| gzip -d \| python ../postprocess.py| ${HADOOP_HOME}/bin/hadoop fs -D hadoop.job.ugi="username,pwd" -copyFromLocal - ${hdfs_outfile}

Python版

import osinput_file = os.environ['mapreduce_map_input_file']
#do something else

References

Hadoop Streaming相关官方文档：https://hadoop.apache.org/docs/r3.1.2/hadoop-streaming/HadoopStreaming.html
Hadoop Streaming入门：http://icejoywoo.github.io/2015/09/28/introduction-to-hadoop-streaming.html
Hadoop排序工具用法小结：http://www.dreamingfish123.info/?p=1102
Hadoop压缩选项权衡：https://www.slideshare.net/Hadoop_Summit/singh-kamat-june27425pmroom210c