电子书《Hadoop数据分析》摘录

大数据基础概念

大数据定义与特征

大数据（Big Data）是指在信息化时代产生的海量数据，来源广泛，包括社交媒体、传感器、IoT 设备、移动设备、网站、数据库等。

五大特征（5V）：

数据产品

数据是编程界的"货币"。数据产品是数据与推断或预测的统计算法的结合，是一个从数据中学习、自适应并且广泛适用的系统。

数据产品示例：推荐系统、恒温器、无人驾驶汽车、智能电网

分析工作流

典型分析工作流：采集 → 整理 → 建模/计算 → 报告和可视化

大数据流水线：包括采集、分段、计算和工作流管理四个主要阶段的迭代模型。工作流管理输出的有用信息还可能通过反馈流机制作为下一次迭代的输入。

数据团队角色

• 数据工程师
• 数据分析师
• 领域专家

Hadoop 生态系统

Hadoop 定位

Hadoop 是大数据操作系统，包含以下核心组件：

核心工作原理

• 数据添加到集群后会被分发到各个节点
• 数据块大小一般为 128MB，跨系统多次复制以提供冗余
• 一个大作业被分解为小任务
• 开发人员专注于数据和计算，不用担心底层的分布式编程细节

Hadoop 集群架构

集群概念

Hadoop 集群是运行 HDFS 和 YARN 的一组计算机，每台就是一个节点。

Hadoop 进程是服务，意味着它们一直在集群节点上运行，接受输入并通过网络传递输出。

一般来说，拥有 20~30 个 worker 节点和单个 master 节点的集群足以在几十太字节的数据集上同时运行多个作业。

节点类型

HDFS 详解

基本架构

HDFS 采用集中式存储架构，一般用于存储原始输入数据或者计算阶段的中间结果。

文件块机制

• 大文件拆分为块并分发给其他计算机
• 每个块都将复制三份（实际可用磁盘空间比硬件磁盘空间少）
• NameNode 记录块所在的位置，定位块的位置
• Secondary NameNode 执行内务任务，合并日志

HDFS 服务角色

伪分布式模式

在"伪分布式模式"中，单个机器将运行所有 Hadoop 守护进程，就好像它是集群的一部分，但网络流量是通过本地环回网络接口流动的。

这种模式虽然没有发挥出分布式架构的优势，但却是一种完美的开发模式，因为不必为管理多台机器而费心。Hadoop 开发人员通常使用伪分布式环境，该环境通常位于虚拟机内部，通过 SSH 连接虚拟机。

数据湖泊

“数据湖泊"中的数据通过一次写入可以多次读取，并创建大量的异构数据用于不同的计算分析。

YARN 详解

引入背景

Hadoop 2 时引入 YARN，突破了 MapReduce 作业/工作负载管理功能与集群/资源管理功能高度耦合的限制，将工作负载管理与资源管理分离。

YARN 服务角色

MapReduce 编程模型

基本概念

Python 中 map 和 reduce 函数的伪代码：

1
2
3
4
5
6
7

def map(key, value):
    # 执行映射操作
    return (intermed_key, intermed_value)

def reduce(intermed_key, values):
    # 执行聚合操作
    return (key, output)

执行流程

1. mapper 操作的副本分发到集群上的每一个节点
2. HDFS 的本地数据以键值对的形式被加载到一个映射过程
3. mapper 输出零个或多个键值对，将计算所得的值映射到一个特定的键上
4. map 和 reduce 之间需要 shuffle 和 sort 进行协调
5. 基于键对键值对进行 sort 和 shuffle 操作，然后将它们传递给 reducer
6. reducer 输出零个或多个最终的键值对

单词计数示例

单词计数应用程序以一个或多个文本文件作为输入，生成一份单词及其频率的列表。

1
2
3
4
5
6
7

def map(dockey, line):
    for word in line.split():
        emit(word, 1)

def reduce(word, values):
    count = sum(value for value in values)
    emit(word, count)

emit 是一个执行 Hadoop I/O 的函数，类似于 Python 中的 yield 函数。

社交网络分析示例

查看用户间有哪些共同好友：

1
2
3
4
5
6
7
8
9

def map(person, friends):
    for friend in friends.split(","):
        pair = sort([person, friend])
        emit(pair, friends)

def reduce(pair, friends):
    shared = set(friends[0])
    shared = shared.intersection(friends[1])
    emit(pair, shared)

JAR 文件执行

MapReduce 的 API 是用 Java 编写的，因此提交给集群的 MapReduce 作业是编译好的 Java 归档（Java Archive，JAR）文件。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 编译
hadoop com.sun.tools.javac.Main WordCount.java
jar cf wc.jar WordCount*.class

# 提交作业
hadoop jar wc.jar WordCount shakespeare.txt wordcounts

# 查看结果
hadoop fs -ls wordcounts
hadoop fs -cat wordcounts/part-00000 | less

# 作业管理
hadoop job -list
hadoop job -kill $JOBID

Hadoop Streaming

概念说明

Hadoop Streaming 是一个 JAR 文件的实用程序，可以指定输入输出的 HDFS 路径参数，还可以指定 mapper、reducer 的可执行程序。

注意：这里的 Streaming 是指标准的 Unix 流（stdin、stdout、stderr）。

Python 实现示例

mapper.py：

1
2
3
4
5
6
7

#!/usr/bin/env python
import sys

if __name__ == "__main__":
    for line in sys.stdin:
        for word in line.split():
            sys.stdout.write("{}\t1\n".format(word))

从 sys.stdin 读取每一行，使用空格拆分，然后逐行将得到的每个单词和数字 1 写入 sys.stdout。

reducer.py：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

#!/usr/bin/env python
import sys

if __name__ == '__main__':
    curkey = None
    total = 0
    for line in sys.stdin:
        key, val = line.split("\t")
        val = int(val)
        if key == curkey:
            total += val
        else:
            if curkey is not None:
                sys.stdout.write("{}\t{}\n".format(curkey, total))
        curkey = key
        total = val

注意：stdin 是接收从 mapper 输出的、经过排序的逐行输入，故可以如上编程实现 wordcount。

CSV 数据处理示例

mapper.py：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

#!/usr/bin/env python
import sys
import csv
SEP = "\t"

class Mapper(object):
    def __init__(self, stream, sep=SEP):
        self.stream = stream
        self.sep = sep

    def emit(self, key, value):
        sys.stdout.write("{}{}{}\n".format(key, self.sep, value))

    def map(self):
        self.status("mapping started")
        for row in self:
            if row[6] < 0:
                self.counter("early departure")
            else:
                self.counter("late departure")
            self.emit(row[3], row[6])  # 始发机场作为键，起飞延误时间作为值

    def __iter__(self):
        reader = csv.reader(self.stream)
        for row in reader:
            yield row

    def status(self, message):
        sys.stderr.write("reporter:status:{}\n".format(message))

    def counter(self, counter, amount=1, group="ApplicationCounter"):
        sys.stderr.write("reporter:counter:{},{},{}\n".format(group, counter, amount))

if __name__ == '__main__':
    mapper = Mapper(sys.stdin)
    mapper.map()

reducer.py：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

#!/usr/bin/env python
import sys
from itertools import groupby
from operator import itemgetter
SEP = "\t"

class Reducer(object):
    def __init__(self, stream, sep=SEP):
        self.stream = stream
        self.sep = sep

    def emit(self, key, value):
        sys.stdout.write("{}{}{}\n".format(key, self.sep, value))

    def reduce(self):
        for current, group in groupby(self, itemgetter(0)):
            self.status("reducing airport {}".format(current))
            total = 0
            count = 0
            for item in group:
                total += item[1]
                count += 1
                self.counter("airports")
            self.emit(current, float(total) / float(count))

    def __iter__(self):
        for line in self.stream:
            try:
                parts = line.split(self.sep)
                yield parts[0], float(parts[1])
            except:
                continue

if __name__ == '__main__':
    reducer = Reducer(sys.stdin)
    reducer.reduce()

本地测试与集群执行

本地测试：

1
2
3

chmod +x mapper.py
chmod +x reducer.py
cat flights.csv | ./mapper.py | sort | ./reducer.py

集群执行：

1
2
3
4
5
6
7

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
    -input flights.csv \
    -output average_delay \
    -mapper mapper.py \
    -reducer reducer.py \
    -file mapper.py \
    -file reducer.py

Windows 系统注意事项：

1
2
3
4
5
6
7
8

hadoop jar D:\hadoop\hadoop-3.2.3\share\hadoop\tools\lib\hadoop-streaming-*.jar
    -D stream.non.zero.exit.is.failure=false
    -files file:///D:/jupyterproject/hadoop/mapper.py
    -files file:///D:/jupyterproject/hadoop/reducer.py
    -input file:///D:/jupyterproject/hadoop/simpletest.csv
    -output /user/simpletest/output
    -mapper "D:\python\python.exe mapper.py"
    -reducer "D:\python\python.exe reducer.py"

Combiner 与 Partitioner

Combiner：因为 mapper 会产生大量的中间数据，需要经过网络传输进行 shuffle 和 sort，而网络资源有限。Combiner 本质是在数据传输之前做一个局部的 reduce 来减少中间数据，预先计算每个键的和，减少生成的键值对数量，从而减少网络流量。

Partitioner：通过划分键空间将键值分配给每个 reducer，分配会尽量均匀以平衡各 reducer 的工作负载。

工作链

工作链可以解决复杂的算法，将其分解为几个较小的 MapReduce 任务。开发人员必须考虑每一步计算怎样 reduce 出中间值。

Pearson 相关系数计算示例：

• 第一个 MapReduce 作业计算 n 以及 x 和 y 的平均值

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

  class VariablePairsMapper(Mapper): def map(self): # 计算(x, y)的个数及和 # 输出键是x在vector中的索引 for y, vector in self: for x, i in enumerate(vector): self.emit(i, (1, x, y) ) class PairsMeanReducer(Reducer): def reduce(self): for key, values in self: # 将所有的值加载到内存，这样可以迭代两次 values = list(values) # 计算(x, y)的和及元素的个数 sx, sy, sn = 0 for (n, x, y) in values: sn += n sx += x sy += y # 计算x和y的平均值 xbar = sx / n ybar = sy / n # 发射每一个(x, y)及x和y的平均值 for (n, x, y) in values: self.emit(key, (x, y, xbar, ybar))

• 第二个 MapReduce 作业计算协方差和标准差，将第一个作业的输出作为输入

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

  import math class PearsonMapper(Mapper): def map(self): # 计算x和xbar的差及y和ybar的差 # 发射差的乘积及其平方 for i, (x, y, xbar, ybar) in self: xdiff = x-xbar ydiff = y-ybar self.emit(i, (xdiff*ydiff, xdiff**2, ydiff**2)) class PearsonReducer(Reducer): def reduce(self): for key, values in self: # 计算差乘积的和以及平方的和 sxyd = 0 sxd2 = 0 syd2 = 0 for (xyd, x2d, y2d) in values: sxyd += xyd sxd2 += x2d syd2 += y2d # 发射相关系数 r = sxyd / (math.sqrt(sxd2) * math.sqrt(xyd2)) self.emit(key, r)

Spark 内存计算

RDD 概念

RDD（Resilient Distributed Dataset，弹性分布式数据集）是一个数据模型，可以根据转换过程重建、从分布式存储读取写入、高速缓存在 worker 节点的内存中以快速重用。

Spark 特点

Spark 主要通过 RDD 实现高速运行。该数据模型在计算时存储在内存中，从而避免了昂贵的中间内存计算磁盘写操作。

Spark 程序员不是简单地指定 map 步骤和 reduce 步骤，而是在执行某个需要协调的动作（如规约或写入磁盘）之前，指定一系列将应用于输入数据的数据流转换。

Spark 栈

Spark 是一个计算框架，专注于计算而非数据存储，可以利用 YARN 和 HDFS 进行集群管理和分布式存储。

主要组件：

• Spark SQL
• Spark Streaming
• MLlib
• GraphX

分布式分析模式

数据流与 DAG

单个运算只对数据进行多个微小的处理，而要想得到更有意义的结果，必须将这些运算组成一个被称为数据流的分步序列。数据流被描述为有向无环图（DAG）。

DAG（有向无环图）用于描述数据流步骤，步骤相连且单步不重复。

限制：许多算法不能轻易转换为 DAG：

• 需要维持或更新单个数据结构的算法（需要共享内存）
• 依赖中间步骤计算结果的算法（需要中间进程间通信）
• 引入循环的算法，特别是循环次数不限的迭代算法

键计算模式

Web 日志记录是 Hadoop 大数据计算的典型数据源，因为它们代表每个用户的点击流式数据。

复合键处理：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import re
from datetime import datetime

dtfmt = "%d/%b/%Y:%H:%M:%S %z"
linre = re.compile(r'^(\w+) \- \- \[(.+)\] "(.+)" (\d+) ([\d\-]+)$')

def parse(line):
    match = linre.match(line)
    if match is not None:
        parts = match.groups()
        date = datetime.strptime(parts[1], dtfmt).timetuple()
        return (parts[0], date.tm_year, date.tm_yday)

序列化与反序列化

• 序列化：将内存中的对象转换成字节流，使其可以被写入磁盘或通过网络传输
• 反序列化：相反的过程
• 简单地使用内置的 str 函数对不可变类型（如元组）进行序列化

键空间模式

Hadoop 环境配置

系统准备

1
2

sudo apt-get update && sudo apt-get upgrade
sudo apt install vim net-tools openssh-server

用户创建

1
2
3

sudo addgroup hadoop
sudo adduser --ingroup hadoop hadoop
sudo usermod -a -G hadoop student

SSH 配置

1
2
3

ssh-keygen
cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys
chmod 600 ~/.ssh/authorized_keys

Java 安装

1

sudo apt install openjdk-8-jdk-headless

配置环境变量（~/.bashrc）：

1
2

export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
export JRE_HOME=/usr/lib/jvm/java-8-openjdk-amd64/jre

IPv6 禁用

Hadoop 主要在 IPv4 网络上测试和开发，对 IPv6 支持不完善，需要禁用 IPv6 防止端口绑定问题。

编辑 /etc/sysctl.conf：

1
2
3

net.ipv6.conf.all.disable_ipv6 = 1
net.ipv6.conf.default.disable_ipv6 = 1
net.ipv6.conf.lo.disable_ipv6 = 1

验证：cat /proc/sys/net/ipv6/conf/all/disable_ipv6（输出 1 表示成功）

Hadoop 安装

1
2

curl -O http://apache.mirror.com/hadoop-2.5.0/hadoop-2.5.0.tar.gz
# 解压、配置四个文件、生成 log 文件地址、启动服务

HDFS 常用操作

-l 参数会列出权限、副本数、用户、组、以字节为单位的文件大小、最后修改日期和时间，以及文件名。

技术对比：POSIX 与 Java 虚拟机

POSIX 标准

POSIX（Portable Operating System Interface of UNIX）是可移植操作系统接口，定义了操作系统应该为应用程序提供的接口标准，旨在获得软件可移植性。

Unix 设计哲学：“舍高效率而取可移植性”

优点：

• 定义了一套操作系统服务的接口，接近系统底层，通常可实现高性能程序
• 不需要中间层支持，带来强大的跨平台可移植能力

现实：很多兼容 POSIX 标准的操作系统实际上是在自身原有 API 基础上封装创建 POSIX 兼容层，二次封装会占用多一些系统资源。

这很容易让人想到另一种跨平台解决方案，例如 Java 虚拟机

Java 虚拟机

可以说它的跨平台能力是 靠牺牲性能 而换取来的
它通过在宿主操作系统上 模拟一个运行时环境 来实现跨平台
这种方法确实可能会牺牲一些性能，因为所有的操作都需要通过 虚拟机这个中间层 来执行
但是，它也提供了一些重要的优点，比如：
虚拟机提供了一个隔离的环境，应用程序不会影响到宿主操作系统或其他应用程序
虚拟机可以提供额外的安全层，限制应用程序的权限
使得应用程序的易于部署，因为它们包含了运行应用程序所需的所有依赖

而遵守 POSIX 标准开发的程序在支持 POSIX 标准的操作系统间运行是不需要依靠类似虚拟机这种中间层的支持的
这就能够在不损失性能的前提下，带来强大的跨平台可移植能力
当然这种理解很美好，但实际上很多兼容 POSIX 标准的操作系统所做的实现是：
在自身原有的 API 接口的基础之上再封装创建一层 POSIX 兼容层来提供对 POSIX 支持
二次封装意味着会占用多一些的系统资源

后续学习方向

拓展与后续学习建议

1. 深入 Spark 生态：学习 Spark SQL、DataFrame、Dataset API，以及 Spark MLlib 机器学习库的使用
2. 流式计算：了解 Kafka + Spark Streaming 或 Flink 的实时数据处理架构
3. 数据仓库：学习 Hive 数据仓库的优化、HiveQL 以及与 HBase 的集成
4. 集群管理：了解 Hadoop 集群的规划、部署和高可用（HA）方案
5. 性能调优：学习 JMX 监控、GC 调优、内存配置优化等性能优化技巧
6. 安全机制：学习 Hadoop 的安全体系，包括 Kerberos 认证、HDFS 权限控制、加密存储等