大数据hadoop核心之 HDFS和MapReduce_hadoop中的hdfs和mapreduce

一.Hadoop简介

• 开源的大数据框架
• 分布式计算解决方案
• HDFS+MapReduce

1.1 hadoop核心之HDFS(分布式文件系统)

• 概念
  • Block 数据块: 存储单元: 默认大小:64M, 一般设置为128M, 备份x3
  • NameNode 名称节点: 管理文件系统的命令空间,存放元数据
    • secondry NameNode: 是NameNode的备份
  • DataNode 数据节点:
• 优点
  • 适合大文件存储, 支持TB\ PB级的数据存储,并有副本策略
  • 可以构建在廉价的机器上, 并有一定的容错和恢复机制
  • 支持流式数据访问, 一次写入, 多次读取最高效
• 缺点
  • 不适合大量小文件存储
  • 不适合并发写入, 不支持文件随机修改
  • 不适合随机读等低延时的访问方式
• 使用shell对HDFS进行操作的常用命令
  • ./hdfs dfs -help: 获取命令帮助
  • hdfs dfs -copyFromLocal /home/hadoop/mk.txt /test/: 上传文件
  • hdfs dfs -copyToLocal: 下载文件
  • hdfs -get: 下载文件
  • hdfs -put: 上传文件
  • jps: 检查hadoop是否已经启动
• 使用python 操作HDFS,具体参考hdfs3 API文档

from hdfs3 import HDFileSystem
host = 'localhost'
port = 9000
client = HDFileSystem(host='localhost', port=9000)
1
2
3
4

1.2 hadoop核心之MapReduce(分布式并行计算系统)

• YARN (2.0之后的资源管理器)
  • ResourceManager
    • 分配和调度资源
    • 启动并监控ApplicationMaster
    • 监控NodeManager
  • ApplicationMaster
    • 为MR类型的程序申请资源, 并分配给内部任务
    • 负责数据的切分
    • 监控任务的执行及容错
  • NodeManger
    • 管理单个节点的资源
    • 处理来自ResourceManager的命令
    • 处理来自ApplicationMaster的命令
• MapRdeuce(python示例)
  • 在ipython中进行简单的示例

$ ipython
Python 3.7.3 (default, Mar 28 2019, 10:38:38) [MSC v.1915 32 bit (Intel)]
Type 'copyright', 'credits' or 'license' for more information
IPython 7.4.0 -- An enhanced Interactive Python. Type '?' for help.

In [1]: from functools import reduce

In [2]: l = ['a', 'bb', 'ccc']

In [3]: l_count = map(len, l)

In [4]: l_count
Out[4]: <map at 0x4a06b70>

In [5]: l_sum = reduce(lambda x, y: x+y, l_count)

In [6]: l_sum
Out[6]: 6

In [7]:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

- 使用python脚本演示
1

# hdfs_map.py
import sys

def read_input(file):
	for line in file:
		yield line.split()

def main():
	data = read_input(sys.stdin)
	for words in data:
		for word in words:
			print('%s%s%d' % (word, '\t', 1))

if __name__ == '__main__':
	main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# hdfs_reduce.py
import sys
from operator import itemgetter
from itertools import groupby


def read_mapper_output(file, separator='\t'):
	for line in file:
		yield line.rstrip().split(separator, 1)

def main():
	data = read_mapper_output(sys.stdin)
	for word, group in groupby(data, itemgetter(0)):
		total_count = sum(int(count) for word, count in group)
		print('%s%s%d' % (word, '\t', total_count))

if __name__ == '__main__':
	main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

将两个python脚本上传到hadoop框架中运行

/usr/local/hadoop/bin/hadoop \
jar /usr/local/hadoop/share/hadoop/tools/lib/hadoop-streaming-2.7.2.jar \
-files "hdfs_map.py, hdfs_reduce.py" \  # 将map和reduce两个处理逻辑上传
-input /test/mk.txt    # 计划进行统计的文档内容
-output /tmp/wordcounttest   # 结果输出的目录
-mapper "python3 hdfs_map.py"  # mapper 处理逻辑
-reduce "python3 hdfs_reduce.py" # reduce 处理逻辑
1
2
3
4
5
6
7