【hadoop】hadoop streaming

API：
https://hadoop.apache.org/docs/stable/hadoop-streaming/HadoopStreaming.html（hadoop3）
https://cwiki.apache.org/confluence/display/HADOOP2/HadoopStreaming（hadoop2）
hadoop version查看hadoop版本，下面的测试基于hadoop2环境。

1.介绍

hadoop streaming可以将任何可执行的文件，如sh脚本，py脚本，嵌入到分布式环境当中执行MR的逻辑，而不必局限于java语言。

hadoop会在每个启动的任务进程中初始化指定的map或reduce脚本并执行，map或reduce任务通过标准输入流读取数据，标准输出流写出数据。

需要注意的是streaming任务默认在map端数据shuffle到reduce端时没有本地合并的过程，也就是MR任务中在map=>reduce过程中，传输的数据格式：key: [value1, value2, value3] ，是key+一个value构成的迭代器。而streaming任务中，map输出的相同key的数据只会被依次相邻的送到同一个reduce，key1: value1; key2: value2这样。

2.Demo

对于下面的数据：

(base) map@gzdt-map-poi-yingxiang-offline04 test_streaming$ cat test.dat
1	小王
2	小李
1	张三
3	李四

执行下面命令：

#!/bin/bash
set -e -x -u -o pipefailhadoop streaming \-input afs://xxx/test.dat \-output afs://xxx/test.res \-jobconf hadoop.job.ugi="xxx,xxx" \-jobconf mapred.job.queue.name="xxx" \-jobconf mapred.job.tracker="xxx" \-inputformat org.apache.hadoop.mapred.TextInputFormat \-mapper cat \-reducer wc

产出的结果文件：

(base) map@gzdt-map-poi-yingxiang-offline04 test_streaming$ cat test.res2       4      181       2       91       2       9

wc是Unix/Linux中的一个命令行工具，用于计算给定文件中的字节数、字数和行数。

wc命令的输出有三个或四个数字：

第一个数字是文件中的行数。
第二个数字是文件中的词数。这里的"词"是指由空格、制表符或换行符分隔的字符串。
第三个数字是文件中的字节数。这是文件的大小，不是字符数。如果文件中包含多字节字符(如UTF-8编码的非ASCII字符)，字节数会大于字符数。
第四个字段是输入文件的名称。

如果只输入wc命令而不带任何文件名，wc会从标准输入读取数据，此时不会显示文件名。

3.map任务获取数据来源

一个streaming job中可以指定多个不同的输入路径，不同路径的数据可能会需要不同的处理方式，所以map任务中区分当前数据的来源非常重要。

hadoop在任务启动时会预置一些属性作为进程级别的环境变量：

圈起来的这个属性在map任务中可以用来获取当前map任务处理的文件块的名字，类似于hdfs://namenode:port/file_path/block_name。

注意点：
1）During the execution of a streaming job, the names of the “mapreduce” parameters are transformed. The dots ( . ) become underscores ( _ )，上面截图中的mapreduce参数值在程序中获取时，需要将.转换为_，也就是mapreduce.map.input.file在程序中通过mapreduce_map_input_file获取。
2）python中该参数被简化，mapreduce_map_input_file写为map_input_file。

下面是一个参考的demo：

import sys
import os
import io# 强制stdout以UTF-8编码输出（python3默认，但最终编码方式受限系统环境）
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')# 当前数据块的全路径
path = os.environ['map_input_file']# path是block的全路径，所以判断全路径包含哪个表名即可
if 'tb1_name' in path:for line in sys.stdin.buffer:decoded_line = line.decode('utf-8').strip('\n')items = decoded_line.split('\t')key = items[0]value = items[1:]print('\t'.join([key, 'tb1_name', value]))  # reduce中通过第二个标志位标识数据来源于哪个表
elif 'tb2_name' in path:for line in sys.stdin.buffer:decoded_line = line.decode('gb18030').strip('\n')items = decoded_line.split('\t')key = items[0]value = items[1:]print('\t'.join([key, 'tb2_name', value]))
else:for line in sys.stdin.buffer:decoded_line = line.decode('utf-8').strip('\n')items = decoded_line.split('\t')key = items[0]value = items[1:]print('\t'.join([key, 'other', value]))

reduce处理：

import sysdef process(datas):passif __name__ == '__main__':key = ''datas = []for line in sys.stdin:items = line.strip('\n').split('\t')if key == '' or items[0] == key:  # 相同bidkey = items[0]datas.append(items)else:process(datas)  # 处理上一批biddatas.clear()  # 清空key = items[0]datas.append(items)  # 添加当前记录if len(datas) > 0:  # 处理最后一批bidprocess(datas)

4.作业提交主要参数解释

1）输入输出

-input：输入文件路径，可以有多个。
-output：输出文件路径。
-inputformat：负责定义如何读取输入数据，并决定如何将数据分割成多个块，每个块由一个map任务处理。默认org.apache.hadoop.mapred.TextInputFormat普通文本读取。

2）资源分发

-file：将本地文件分发到hadoop集群的所有MR任务的当前工作目录（task级别），使任务脚本能够访问这些文件。当前工作目录表明每启动一个map或reduce任务都会拷贝一份副本。常用来分发map和reduce的任务脚本以及一些资源文件，如模型权重文件。模型资源文件和任务脚本处于同一个目录，任务脚本中可以直接./xxx相对路径读取资源文件。
-cacheArchive：将归档文件分发到所有任务节点（worker级别）并自动解压，节点上的MR任务启动时不会拷贝副本到自己的工作目录，会通过符号链接共享同一个节点上的文件。适合用来分发一些比较大的文件，如python环境包。

eg：-cacheArchive $HADOOP_PYTHON3#python3，如果hdfs上python包解压缩后的结构为python3/bin/python3，则在mapper和reducer参数中指定python环境时-mapper ./python3/python3/bin/python3 mapper.py，关于python解释器层级路径的解释：#python3，会在当前task工作目录中创建一个名为python3的文件夹，并将python环境的tar包拉到这个目录下解压。

3） 作业启动管理

-mapper： 指定mapper脚本。
-reducer： 指定reducer脚本。
-jobconf mapred.map.tasks：map任务数量，不严格依赖指定，跟输入数据的分块有关。
-jobconf mapred.reduce.tasks：reduce任务数量，等于任务最终输出包含的文件个数。
-jobconf hadoop.job.ugi：提交作业用户的身份，主要影响作业对HDFS或YARN的访问权限。
-jobconf mapred.job.queue.name：资源队列。
-jobconf mapred.job.priority：任务优先级。

4）分区规则

-partitioner：分区规则，org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner基于指定序号的字段为分区键。
-jobconf stream.num.map.output.key.fields：指定map输出的key，如-jobconf stream.num.map.output.key.fields=1指定按照map输出的第一个字段作为key。
-jobconf num.key.fields.for.partition：决定如何分区，默认按照完整的key，即上面参数配置。如果上面配置为符合key，该参数可以指定按照key中的部分字段分区。也就是说，如果配置了这个参数，指定的字段决定了map输出送往哪个reduce，而上面配置的key决定了送到同一个reduce的数据的先后顺序。(map在shuffle数据到reduce之前会对数据按照key排序)

5）压缩

-jobconf mapred.output.compress：输出结果是否开启压缩，true、false。
-jobconf mapred.output.compression.codec：指定压缩编解码器，Gzip压缩org.apache.hadoop.io.compress.GzipCodec。
-jobconf mapred.output.compression.type：压缩类型，=BLOCK块压缩，适合列式存储，=RECORD记录压缩，适合行式存储。

5.仅map作业

https://hadoop.apache.org/docs/stable/hadoop-streaming/HadoopStreaming.html#Specifying_Map-Only_Jobs
在hadoop3版本中指定参数-D mapreduce.job.reduces=0，hadoop2版本中指定-jobconf mapred.reduce.tasks=0。

-reducer参数不可省，即使不需要reduce任务，否则实际测试中会报错如下。可以随意指定，例如-reducer cat。