数据工程指南高级技能：处理框架

在之前的文章中，我们探讨了数据工程的入门知识和基础技能，接下来继续探讨高级技能。

数据工程指南系列：

《数据工程指南：初学者入门》
《数据工程指南：基础技能》
《数据工程指南高级技能：如何选择数据平台》
《数据工程指南高级技能：Hadoop平台》
《数据工程指南高级技能：连接和缓存》

原文来源于Github开源项目《The Data Engineering Cookback》
作者：Andreas Kretz
感兴趣可以查看：Data Engineering Cookbook

Lambda和Kappa架构

在这个视频中，我们讨论了带流处理和批处理的lambda架构，以及只包含流处理的Kappa架构。

批处理(Batch Processing)

还记得上一年的纳税申报表吗？

你把文件夹拆开，然后在房子里跑来跑去找收据。

当终于找到了所有的东西，填好表格，然后把它寄过去。

做税务报表是批处理过程的一个主要例子。

获取数据，存储下来，然后加载数据并分析，获得见解：

批处理可以是定时任务，也可以是一次性任务。我们先提出一个宏观的问题，通过批处理来观察全局并获得深刻见解。

因此，批处理需要使用大量数据，这些数据一般存储在类似Hadoop HDFS的系统中。

它们可以轻松地存储PB级别的数据。

批处理作业的结果非常有用，但执行时间很长，因为处理的数据量很大，往往需要几分钟甚至几小时才能得到结果。

流处理(Stream Processing)

实时深入了解您的数据。

流媒体允许用户根据“实时”分析快速做出决策和采取行动。

与批处理相反，流处理是在获得数据的瞬间进行分析。

使用流处理，不必等待几分钟或几个小时才能得到结果，您可以实时了解数据。

在批处理流程中，分析是在数据存储之后进行的，它可以访问所有可用的数据。

流处理在数据存储之前进行分析，它只能访问传入的数据片段。

因此，分析洞察的能力是受限的，因为无法获取全局数据。

只有通过流媒体技术，您才能为客户创建高级服务。例如，网飞(Netflix)将流处理合并到Chuchwa V2.0和新的Keystone管道中。

通过流处理实现高级服务的一个例子是Netflix“Trending Now”功能。感兴趣的小伙伴可以去了解一下。

三种流处理方法

在处理消息时，流处理有三种方法：至少处理一次，至多处理一次，仅处理一次。在创建解决方案时，将它们牢记在心非常重要。

至少处理一次

至少一次，意味着消息在系统中被处理一次或多次。因此，至少有一次消息进入系统并且没有被处理是不可能的。

它不会在系统中被删除或丢失。

考虑汽车管理，您可以从汽车上获取GPS数据，该数据包含时间戳和GPS坐标。重要的是，你至少得到一次GPS数据，这样你就知道车在哪里，由于有时间戳，它被多次处理并不重要，或者它会被存储多次，因为它会覆盖现有的数据。

最多处理一次

顾名思义，处理消息的次数不能超过1次，这意味着可以删除一些信息。

一个例子是事件处理(event processing)。有些事件正在发生，但该事件不重要，因此可以删除它，删除后也不会产生任何后果。

但是当目标事件发生时，最多被处理一次，否则就会产生目标事件发生了很多次的假象。

仅处理一次

有的消息只能被处理一次。

这方面的一个例子是银行业，当你考虑信用卡交易时，放弃交易是不好的。

如果你的钱被退回，你的付款就不会过去了。多次处理事务也不好，这意味着要支付多次费用。

流处理还是批处理？

从批处理开始是个好主意，批量处理是每个大数据平台的基础。

批处理体系结构很简单，因此设置起来很快。平台的简单性意味着，它的运行成本也相对便宜。

一个批处理平台可以让你快速回答宏观问题，为客户提供宝贵见解。

随着业务发展，还需要做动态分析，然后添加一个流媒体管道到你的批处理大数据平台。

ETL过时了吗？

ETL在数据科学和大数据中过时了吗？在今天的播客中，我将分享对于ETL（提取、转换、加载）问题的看法，包括ETL是否仍在使用，或者预处理和清理取代了它，在数据工程中什么会取代ETL。

MapReduce

从Hadoop生态系统的早期开始，MapReduce框架是Hadoop和HDFS的主要组件之一。

例如，Google使用MapReduce来分析网站存储的HTML内容，通过计算所有HTML标签、所有单词和它们的组合（例如标题）。输出结果随后被用于创建Google搜索的页面排名。

那时每个人都开始为谷歌搜索优化自己的网站，搜索引擎优化诞生了，那是2004年。

MapReduce的工作原理是分两个阶段处理数据：map阶段和reduce阶段。

在映射阶段，框架从HDFS读取数据，每个数据集称为输入记录。

在reduce阶段，进行实际计算并存储结果。存储目标可以是数据库或备份HDFS或其他对象。

MapReduce的魔力在于如何实现map和reduce阶段，以及两个阶段如何协同工作。

map和reduce阶段是并行的。这意味着，您有多个映射阶段（mapper）和reduce阶段（reducer），它们可以在集群机器上并行运行。

下面是一个map和reduce过程如何处理数据的示例：

MapReduce的工作原理是什么？

首先，整个map和reduce过程很大程度上依赖于使用键值对。

在映射阶段，输入数据（例如文件）被加载并转换为键值对。

当每个映射阶段完成后，它将创建的键值对发送到reducer，在那里它们将按键进行排序。这意味着，reduce阶段的inputrecord是映射器中所有具有相同键的值的列表。

然后reduce阶段对该键及其值进行计算并输出结果。

可以并行使用多少个映射器和还原器？并行映射和reduce进程的数量取决于集群中有多少CPU核，每个映射器和每个减速器都使用一个核心。

这意味着你拥有的CPU核越多，可以使用的映射器越多，提取过程就越快。你使用的减速器越多，实际计算的速度就越快。

为了更清楚地说明这一点，我准备了一个例子：

案例

正如我之前所说，MapReduce分两个阶段工作，map和reduce。通常用单词计数(word count)任务来解释这个过程。

我个人比较讨厌单词计数的例子，因为计算东西太琐碎了，并不能真正地展示使用MapReduce可以做什么。因此，我们将使用物联网世界中更真实的案例。

物联网应用程序创建了大量必须处理的数据。这些数据是由物理传感器生成的，例如测量8点钟的室温。

每次测量都由一个键（测量时的时间戳）和一个值（传感器测量的实际值）组成。

因为您的机器上通常有多个传感器，或者连接到您的系统，所以密钥必须是复合密钥。除测量时间外，复合键还包含有关信号源的测量时间信息。

为了简单起见，让我们暂时忘掉复合键。假设我们只有一个传感器，每个度量都输出键值对，例如：Timestamp value。

此练习的目标是创建传感器数据的每日均值。

下图显示了map和reduce过程的工作原理。

首先，映射阶段通过键和值从源（例如HDFS）加载未排序的数据（输入记录）(key：2016-05-01 01:02:03，value：1).

然后，因为目标是获得每日均值，剔除时间戳中包含的时分秒信息。

这就是map阶段发生的一切，仅此而已。

在所有并行映射阶段完成之后，每个键值对都被发送到一个reducer，后者处理这个特定键的所有值。

每个reducer输入记录都有一个值列表，您可以计算（1+5+9）/3，（2+6+7）/3和（3+4+8）/3。

您还可以看到，为什么map reduce对于并行工作如此有用。在这种情况下，map阶段可以由9个平行的Mapper完成，因为映射过程是相互独立的。

reduce阶段仍然可以由三个任务并行完成，一个代表橙色，一个代表蓝色，一个代表绿色。

这意味着，如果你的数据集变大10倍，机器性能提升10倍，那么计算的时间是一样的。

MapReduce的极限在哪里

MapReduce对于简单的分析任务来说非常棒，比如计数。

首先MapReduce将数据从HDFS加载到映射函数，然后在reducer中开始处理数据，最终将结果写入数据仓库。

MapReduce的问题是无法轻松地将多个map和reduce进程链接在一起，在每个reduce过程结束后，数据必须存储在某个地方。

这使得MapReduce无法处理复杂的分析任务，你需要把MapReduce的工作链在一起。

将存储和加载中间结果的作业链接起来毫无意义。

MapReduce的另一个问题是它不能处理流式分析。启动，分析和关闭进程需要一段时间，一般都需要好几分钟。

在实时数据分析越发流行的背景下，这是一个很大的缺点。

Apache Spark

Spark与MapReduce有什么区别？

Spark是一个完整的内存框架，例如将数据从HDFS加载到内存中。

不再有固定的map和reduce阶段，你的代码可以变得非常复杂。

一旦进入内存，输入数据和中间结果将保留在内存中（直到作业完成），它们不会被写入像MapReduce这样的驱动器。

这使得Spark成为进行复杂分析的最佳选择。例如，它允许你执行迭代过程，多次修改数据集以创建想要的输出。

流分析功能也是Spark如此强大的原因，Spark可以每隔几毫秒或几秒执行定时作业。因此Spark可以“实时”地从流数据中传递结果。

Spark如何与Hadoop结合？

有一些非常误导性的文章，标题是“Spark or Hadoop”，“Spark比Hadoop好”甚至“Spark正在替代Hadoop”。

现在是时候让你看看Spark和Hadoop之间的区别，然后你会明白何时以及为什么应该使用Spark和hadoop。

你也会明白为什么“Hadoop还是Spark”是完全错误的问题。

区别在哪里？

为了弄清楚Hadoop与Spark的区别，我创建了一张简单的功能表：

Hadoop用于在分布式文件系统（HDFS）中存储数据，它可以用MapReduce分析存储的数据，用YARN管理资源。

然而，Hadoop不仅仅是存储、分析和资源管理，Hadoop核心有一个完整的工具生态系统。

与Hadoop相比，Spark只是一个分析框架，它没有存储功能，虽然它有一个独立的资源管理，但基本上很少使用。

Spark和Hadoop完美融合

如果Hadoop和Spark不是一回事，它们能一起工作吗？当然可以。

使用HDFS存储，通过Spark进行分析，利用YARN实现资源管理。

为什么两者能很好地配合？

从平台架构的角度来看，Hadoop和Spark通常在同一个集群上管理。这意味着在运行HDFS的每台服务器上，也会运行一个Spark worker线程。

在分布式处理中，机器之间的网络传输是一个很大的瓶颈，在一台机器内传输数据会大大减少这种流量。

Spark能够确定所需数据在哪个数据节点上储存，并将数据从本地存储器直接加载到本机内存。

这大大减少了网络流量。

Spark和YARN

你需要确保你的物理资源在服务之间得到了完美的分配。在同一台机器上运行Spark workers和其他Hadoop服务时尤其如此。

让两个资源管理器管理同一服务器的资源是没有意义的，他们迟早会互相影响。

这就是为什么Spark的独立资源管理器很少被使用。

所以，问题不是选择Spark还是Hadoop，而是你应该在Hadoop的HDFS和YARN之外使用Spark或MapReduce吗？

我的经验总结

如果你正在做一些简单的批处理工作，比如计数或求平均值：使用MapReduce。

如果你需要更复杂的分析，如机器学习或快速流处理：使用Spark。

支持的编程语言

Spark作业可以用多种编程语言实现，这使得创建分析过程对数据科学家非常方便。

Spark支持Python、Scala和Java。在SparkR的帮助下，你甚至可以将你的R程序连接到一个Spark集群。

如果你是一个非常熟悉Python的数据科学家，那就用Python吧。如果你知道如何编写Java代码，我建议你开始使用Scala。

Spark作业在Scala中比在Java中更容易编码。在Scala中，可以使用匿名函数进行处理。

这样可以减少开销，并得到更干净、更简单的代码。

在Java8中，lambda expressions引入了简化的函数调用。不过很多人，包括我在内，更喜欢Scala而不是java。

Spark如何调用Hadoop的数据

我总是强调，在存储数据的地方对数据进行本地处理是最有效的方法。

这正是Spark正在做的。您可以而且应该直接在Hadoop集群的数据节点上运行Spark workers。

然后Spark可以在本机识别存储所需数据的数据节点。这使得Spark能够使用在存储数据的机器上运行的worker将数据加载到内存中。

这种设置的缺点是你需要更昂贵的服务器，因为Spark processing需要更强大的服务器和更多的RAM和CPU。

什么是RDDs，如何使用它们

RDDs是Spark的核心部分，在学习Spark之前我会先学习和使用RDD，它来自MapReduce。现在你使用数据帧或数据集。

我觉得了解RDD和Spark如何在较低级别上工作是很有价值的。

如何使用SparkSQL？

当你使用Apache Zeppelin笔记本学习Spark时，你会遇到SparkSQL。SparkSQL允许您使用类似sql的语法访问数据。

尤其是当你使用notebook时，从你的数据中创建图表是非常方便的。

什么是数据帧，如何使用它们

就像我之前说的，数据帧(dataframes)是RDD的继承者，它是新的Spark API。

数据帧是类似Excel电子表格的存储工具，这使得SparkSQL非常容易使用和操作它们。

使用数据帧处理比使用RDD处理更快，在处理数据时使用了优化算法。

用Spark做机器学习？

Spark上是否能运行基于TensorFlow的深度学习程序？已经有可能了，查看以下链接：

为什么人们要整合Spark和TensorFlow，即使有分布式的TensorFlow框架？

在Spark上运行TensorFlow

使用Spark和TensorFlow实现深度学习

MLlib

机器学习库MLlib包含在Spark中，因此不需要导入另一个库。

我并非数据科学专家，但从我了解到的，机器学习框架mllib对于那些想用Spark训练和应用模型的数据科学家来说是非常不错的选择。

配置Spark

从解决方案架构师的角度来看，Spark非常适合Hadoop大数据平台，这与集群部署和管理有很大关系。

像Cloudera、MapR或Hortonworks这样的公司在其hordoop发行版中加入了Spark，因此可以从Hadoop管理面板中直接部署和管理Spark集群。

Spark资源管理

当运行一个计算框架时，您需要资源来执行计算：CPU，RAM，I/O等。开箱即用的Spark可以使用独立的资源管理器来管理资源。

如果Spark在Hadoop环境中运行，则不必使用Spark自己的独立资源管理器，你可以配置Spark使用Hadoop自带的YARN资源管理器。

为什么要这么做？它允许YARN有效地为Hadoop和Spark进程分配资源。

使用一个而不是两个独立的资源管理器使配置资源更加容易。

Graph DB

图形数据库按节点和关系存储数据。每个节点代表一个实体（人、电影、事物和其他数据点），关系表示节点之间的关系。它们被设计用来存储和处理与数据（在本例中是节点）的重要性相同的关系。这种关系优先的方法有很大的区别，因为不需要再使用外键和主键来推断数据之间的关系。

当应用程序需要在不同数据点之间的多层次关系中导航时，图形数据库尤其有用。

Neo4j

Neo4j是目前最流行的图形数据库管理系统。它兼容ACID，并提供自己的图形数据库实现。除了节点和关系之外，neo4j还包含以下组件，用信息丰富数据模型。

标签(Labels)。它们用于对节点进行分组，并且每个节点都可以分配多个标签。标签被索引以加快在图中查找节点的速度。
属性(Properties)。这些是节点和关系的属性。Neo4j允许将数据存储为键值对，这意味着属性可以有任何值（字符串、数字或布尔值）。

优点

Neo4j无设计模式
高可用性，并提供事务性保证
Cypher是一种声明性查询语言，使图形导航变得非常容易
Neo4j运行速度很快且易于遍历，因为数据是连接的，非常容易查询、检索和导航

缺点

与关系数据库相比，Neo4j不适合任何类型的聚合或排序
并不是处理会计等事务性数据的最佳选择
目前不支持分区

你们的点赞和收藏是我们最大的创作动力，我们每天都会为大家带来数据科学和量化交易领域的精品内容。

蜂鸟数据：开源金融数据接口，一个API连接世界金融市场。

蜂鸟数据团队由业界顶尖的数据工程师，数据科学家和宽客组成，我们正努力构建一个开源的金融数据库，并提供API接口，目标是令金融数据开源化和平民化。

浏览并测试我们接口吧，目前覆盖股票，外汇，商品期货，数字货币和宏观经济领域，包括实时报价(tick)和历史数据(分钟)，提供REST API和Websocket两种接入方式，能够满足金融分析师，量化交易和理财app的需求。

蜂鸟数据API接口文档

登录蜂鸟官网，注册免费获取API密钥