2020-10-07

分布式1007-Map-Reduce的文字流

最后编辑于：20.10.15

开门见山地来一段，就一段，不会有人这个都没搞懂吧，不会吧不会吧（拖走

hadoop jar \
 $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-3.1.3.jar \
 -input /user/devel/2020210995wangyuanhe/README.txt \
 -output /user/devel/2020210995wangyuanhe/1007output \
 -mapper "/usr/bin/cat" \
 -reducer "/usr/bin/wc"

开始前再插一句题外话，被强大而可爱的丰丰老师表（da）扬（shang）了，动力+10086，继续努力啊小禾禾！！

分布式1007-Map-Reduce的文字流

1.程序运行情况介绍

这张运行图只是执行中间一部分，正常情况下无ERROR，map 100% reduce 100%（这里运行时看着最爽）。

图2中第一行文件并无内容，后7个文件是本次运行开启的7个mapper的结果，reducer在这几个文件中运行，并把结果写入这7个文件中，所有的结果求和即真正结果。

下面看看运行的文件情况：

上图中，wc函数第一列的和就是16，即行数（验证正确），第二列为单词数（字符串连在一起算一个单词），第三列为字节数。

单个mapper+单个reducer运行

每次cat：
行数+1；单词+n；字节数+m

服务器上有很多个mapper，本次有17个（见上图），每个程序都做了cat函数（打印），7个reducer一起运行wc（计算行数）。Hadoop jar 中有这样一个参数，num.tasks，控制任务的数量。

2. 运行的相关介绍

reducer结束的很慢，原因是启动时要花资源，map过程非常快（程序运行时有体会）。听说均分文件时会用到哈希code，现在很多算法都是哈希函数的进阶，不知真伪，之后问问。

在传输中，隐含了打乱shuffle和整理sort的过程:

$平摊$

把数据随机打乱，$shuffle$，保证每个mapper接受的任务量相近。
打乱顺序的任务再排序，$sort$，使每个程序尽可能找到较近数据。

由于，数据在HDFS上存储在分布式的硬盘上，必须主动从硬盘读到内存里，有I/O（input/output）的消耗。如果数据很多，读起来很慢。一般map很复杂，可能map的中间结果要写入硬盘，又产生I/O消耗，reduce也需要从硬盘中读取。

故HADOOP对硬件读写的要求很高，如此反过来也节约了内存资源（贵）。真正制约hadoop的大多是硬盘读写，因此很多服务器用SSD，但是SSD很容易坏，故需要做冗余（防止硬件坏掉）。

apply函数，groupby函数，都有map的感觉

3. Hadoop 与 Spark

hadoop擅长进行批处理，但不能进行实时计算（比如无人驾驶）、股票高频交易（短时间的计算），这种实时运算需要使数据保持“热状态”不存入硬盘，在map-reduce后立刻传出，与硬盘无关。

hadoop不擅长，但是spark擅长。spark写入硬盘的操作很有限，因此速度快。当然，上文也提到了，内存比硬盘贵，所以hadoop更廉价，两个框架各有胜负。

同时，hadoop不能实现迭代计算（牛顿迭代，神经网络，梯度下降，反向传播），几乎涵盖所有机器学习算法。迭代时需要大量循环，不能经常读写硬盘。

4. 标准输入输出 STDin & STDout

在计算机编程中，有一类输入输出只与屏幕有关：

Stdout

任何程序结果总是需要保存，但有一类输出直接打印在屏幕上。凡是能打印的都是stdout。举些例子：print函数(r,python)，cat函数(r,linux)，printf函数(c)等等等等。基本全部语言都能标准输出。

Stdin

计算机能够接受打印的“文字流”（这也是hadoop streaming中streaming的含义！），举些例子：如linux和r的管道函数，python里open函数，都是打开文件把每一行读进来。

linux中很方便地组织你的文件，只要文件是文本文件，都可以用管道“吸入”。很多linux的函数都以cat开头（猫猫头）：

1	cat txt \| python wc.py 单机实验

再用R语言举个例子：

sink("想保存的文件名.txt",append = T, splt = T)
abc = c(rnorm(100))
abc
sink()

写进hdfs就是另外一幅模样了：

#! /usr/bin/env Rscript
options(warn=-1)
sink("/dev/null")

input <- file("stdin", "r") # 用input 吸入来自linux的STDin
while(length(currentLine <- readLines(input, n=1, warn=FALSE)) > 0)
{
    fields <- unlist(strsplit(currentLine, ","))
    lowHigh <- c(as.double(fields[3]), as.double(fields[6]))
    stock_mean <- mean(lowHigh)
    sink()
    cat(fields[1], fields[2], stock_mean, "\n", sep="\t")
    sink("/dev/null") # dev/null是linux的黑洞目录，扔进去就会消失呢！
}

close(input)

运行时在linux中用rscript：

1	Rscript test.r

其实做开发时，java语言很好用（类操作，大型项目架构，内存管理），适合创建非常巨大的项目，运行很久不停止，其他语言不行。hadoop也是java编成。但java并非所有人都会，hadoop面向数据处理，会java的人不多。

由此，先辈们做了个能让各种语言都能识别的框架：

hadoop做了很好玩的模块：streaming（标准输入输出的“文本流”）。提供了简单的接口，c，py，java，r……但凡能接受标准输入输出，就可以调用！使得map函数和reduce函数完全脱离了hadoop，只需要输入输出就能得到结果，影响速度的只有map和reduce的写法。

Hadoop不是编程语言，是分布式计算架构。
——李丰老师

5. 我们的函数，部署！

教练，我也想调用hadoop接口跑我自己的程序！

完全没问题！

很简单，首先要保证每个存储数据的节点上（worker节点）必须有函数cat、wc，我们自己写一个wchehe.py，然后放上服务器去就好啦。

如下，就是一个简单的读取行数的py程序，第一行一定要注明函数应该怎么找到运行的地方：

#! /usr/bin/env python3
import sys
linecount=0
data = []
for line in sys.stdin:
truelinecount += 1
truedata.append(line)
Print(linecount)

可以用chmod +x wchehe.py 改一下运行权限。

下来，为了规范代码格式，我们用一个shell批处理文件作为我们的程序入口，也方便调整参数。

开头别忘了告诉sh函数这是个批处理。看到这篇文章的同学不要用原代码直接跑啊（

#! /usr/bin/bash

PWD=$(cd $(dirname $0); pwd)
cd $PWD

HADOOP_inputdir=/user/devel/2020210995wangyuanhe/ordertxtfiles/test-edit.txt
HADOOP_outputdir=/user/devel/2020210995wangyuanhe/output/1007out01
HADOOP_home=/share/hadoop/tools/lib/hadoop-straming-3.1.3.jar

echo $HADOOP_home
echo $HADOOP_inputdir
echo $HADOOP_outputdir

hadoop fs -rm -r $HADOOP_outputdir

hadoop jar \
 $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-3.1.3.jar \
 -files $PWD/wchehe.py \
 -input ${HADOOP_inputdir} \
 -output ${HADOOP_outputdir} \
 -mapper "/usr/bin/cat" \
 -reducer "python wchehe.py"