Hadoop Streaming和awk

1. Hadoop streaming类似于Unix管道数据流，从标准输入(STDIN)输入，输出到标准输出(STDOUT)，数据必须是基于文本的，文本的每一行被认为是一条记录。这也是很多Unix命令，例如：awk的工作方式。在Hadoop streaming中整个数据流就像是一个管道(Pipe)，依次经过mapper，mapper sorter，reducer，Unix伪代码表示如下：

   cat [input_file] | [mapper] | sort | [reducer] > [output]

2. 在streaming中使用Unix命令

        bin/hadoop jar share/hadoop/tools/lib/hadoop-streaming-2.4.0.jar

            -input input/cite75_99.txt -output output  -mapper  'cut -f 2 -d ,'  -reducer 'uniq'

        streaming也可以使用脚本，只要该脚本从STDIN输入，输出到STDOUT

        bin/hadoop jar share/hadoop/tools/lib/hadoop-streaming-2.4.0.jar

            -input input/cite75_99.txt  -output output  -mapper ‘RandomSample.py 10’

            -file RandomSample.py  -D mapred.reduce.tasks=1

 3.  mapper的输出默认以Tab键分割key/value，若没有Tab，则整条记录为key，value为空。Hadoop shuffling将依据key将key/value发送给不同的reducer。但mapper和reducer的工作依然是从STDIN处理一行一行的文本。

    hadoop streaming提供参数-inputformat、-outputformat、-partitioner、-combiner、-numReduceTasks指定相应的hadoop job参数。其中-mapper、-reducer、-combiner既可以指定为类名，也可以指定为unix可执行程序。-file参数指定一个文件，将文件打包，和job一起被发送到各个执行节点。-files参数可以指定多个文件，文件名之间采用逗号分隔。-D参数可以指定键值对型的参数，例如：-D mapred.reduce.tasks=0，-D stream.map.output.field.separator=.，-D stream.num.map.output.key.fields=4，分别指定reduce数量为0，map输出字符串以.分割成字段，并设置前4个字段之前的字符串为map输出的key（即第四个.之前的字符串），第四个.之后的字符串为value。

4.  Hadoop Stream中常结合awk命令使用。

    awk适用于处理字段型数据行，Pattern中可以调用~和！~支持正则表达式匹配。

    awk可以从shell命令获取输入数据，例如命令“ls”| getline。

    awk可以在print和printf语句后使用>和>>将输出重定向到文件，例如：print $1>"myfiles"。

    awk中可以直接调用shell命令，只需将shell命令置入system函数中，例如：system("ls > myfiles")，但是使用system函数时，awk无法直接给shell命令传递参数，shell命令也无法将输出直接传递给awk。

    awk的getline函数也可以使用<运算符，从文件中逐行读取记录。

    可以将awk程序写进一个脚本中，然后调用命令awk -f my.awk input.file执行，也可以直接在shell命令行中执行awk命令，或者将awk命令写进myawk.sh文件执行。

    awk命令以单引号开始，以单引号结束，awk命令中的字符串一律用双引号。

    字段分隔符FS，是awk的内建变量，其默认是空白和\t，可以在BEGIN pattern中指定：FS="[\t,:]+"，表示使用\t、,和:分割字段。

    awk提供close(filename)函数关闭打开的文件，用于确保输出到文件的顺序。

    awk采用内建变量RS，默认为\n分割多行数据。

    awk提供函数split(原字符串，字符串数组，分隔字符)，按照分隔字符将原字符串分割，保存在字符串数组中。

    awk提供内建变量ARGC，表示出了-f、-c等参数之外，命令行参数的个数，通过ARGV，读取命令行参数。

    awk中指定ARGV[1] = "-"，表示由键盘输入数据。

    awk可以使用function关键字定义函数，函数中定义的变量在函数外也是可见的，因此在定义递归函数时，应特别注意。

    awk正则表达式，置入//之间。

    awk提供next指令，跳过当前数据行，处理下一行，提供exit指令，退出awk程序。

    awk提供内建的字符串函数和数学函数，index（原字符串，找寻的字符串）返回找寻的字符串在原字符串中第一次出现的index，length（字符串）计算字符串的长度，match（原字符串，正则表达式）函数返回匹配结果，sprintf函数将格式化输出到字符串，sub（正则表达式，将替换的新字符串，原字符串）将首次匹配正则表达式的部分替代，substr（原字符串，起始位置，长度）返回子字符串；数学函数包括int取整，sqrt求根号，exp，log，sin，cos，rand，srand。
    

    5.  hadoop streaming中使用awk注意事项

    下面是一个awk命令实例：

    hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.2.0.jar \
    -libjars  parquet_test_1.0.jar \
    -input /user/algo/lujianfeng/parquet-mz-input-test08 \
    -output /user/algo/lujianfeng/parquet-mz/countfield10 \
    -mapper "awk -F'[=\^]' '{for(i=1;i<=NF;i++){printf(\"%s\t1\n\",\$i);}}'" \
    -reducer "awk -F'[\t]' 'BEGIN{key=\"\";num=0;}{if(key!=\$1 && length(key)>0)        {printf(\"%s\t%d\n\",key,num);num=0;} key=\$1;num++;}'"
    

    注意：1. awk命令中的$、“字符必须转义

            2. hadoop streaming中的reducer不同于写java MR程序时的reduce函数，所有发送给一个Reducer的Map输出结果，hadoop streaming中都由同一个awk脚本执行，不会按照key进行分组

            3. hadoop streaming中，reducer处理的数据就是map的输出结果，即一行文本

转载于:https://my.oschina.net/jhone/blog/312062