怎么在HDFS中组织和使用数据

这篇文章主要为大家展示了“怎么在HDFS中组织和使用数据”，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下“怎么在HDFS中组织和使用数据”这篇文章吧。

4.1 组织数据

组织数据是使用Hadoop最具挑战性的方面之一。企业中存在来自不同部门和不同人员的压力，例如数据科学家和集群管理员，每个人对数据都有自己的要求。更重要的是，这些要求通常是在数据应用程序投入生产并且已经积累大量数据之后提出。

Hadoop中组织数据有多个维度。首先，我们需要学习如何在HDFS中组织数据，之后将面临实际操作问题，例如对数据进行分区和压缩，决定是否启用Kerberos来保护集群以及管理和传递数据更改。本章目标是关注组织数据过程中一些复杂问题，包括数据分区和压缩，让我们从在HDFS中构建数据开始吧。

4.1.1 目录和文件布局

定义数据组织方式的集群范围标准是一项值得探究的工作，因为它可以更容易地发现数据位置，并且应用和管理可通过数据存储解决的问题。因为我们在文件系统可以表达的范围内工作，所以安排数据的常用方法是创建与企业组织或功能结构一致的层级结构。例如，如果在分析团队工作并且正在将新数据集引入集群，那么组织目录的一种方法将如图4.1所示。

图4.1 HDFS目录布局示例

在学习之前，企业最好已经确定了一种数据格式，例如Avro，这可以随时间推移改进模式。 通过在结构中粘贴版本号，你可以灵活地转移到任何一种新的数据格式，并使用目录路径传达不同的文件格式。

在目录中放置版本号面临的唯一挑战是如何将更改有效传达给数据使用者。如果确实有该问题，企业可以尝试将HCatalog视为从客户端抽象出的一种数据格式。

按日期和其他字段进行分区

你可能会使用目录结构来模拟企业不同部门的数据演变需求，但为什么还需要按日期进一步分区?这是Hive早期使用的一种技术，可以帮助加快查询速度。如果将所有数据放入一个目录中，那么每次访问数据实际都在进行等效的Hadoop全表扫描。相反，根据对数据的访问方式对数据进行分区更为明智。

因为我们事先很难确切知道如何访问数据，但按数据生成日期对数据进行分段是合理的分区尝试。如果数据没有日期，那么请与数据生产者讨论添加日期，因为创建事件或记录的时间是应始终捕获的关键数据点。

4.1.2 数据层

在2012年的Strata演讲中，Eric Sammer提出了对数据存储进行分层的想法，这也很好地与Nathan Marz的Lambda架构的主要原则相关——永远不会删除或修改原始数据。

乍一看，这似乎没有任何意义。因为你只需要提取数据源的重要部分，其余部分可以舍弃，毕竟保留全部原始数据有些浪费，特别是如果有部分未被积极使用，但你很难保证未来坚决不会从未使用的数据中提取到有价值的信息。

我们的软件偶尔也会出现错误。想象一下，如果你正在软件中传输数据，获得想要的结果之后，你决定丢弃源数据，但是你突然发现操作逻辑中存在错误，因为丢弃了源数据将无法返回并重新生成结果。

因此，建议企业根据以下层级考虑数据存储：

• 原始数据是第一层。这是从源捕获的未更改数据，永远不应修改此层数据，因为生成衍生物或聚合的逻辑可能存在错误，如果丢弃原始数据，则无法在发生错误时恢复。

• 第二层：从原始数据创建派生数据。该层，你可以执行重复数据删除和任何其他数据治理要求。

• 第三层：汇总数据。这是根据派生数据计算出来的，可能会被输入HBase系统或选择的NoSQL系统，以便在生产和分析过程中实时访问数据。

数据层也应该在目录布局中有所展示，以便用户可以轻松区分这些层。

一旦确定了用于对数据进行分区的目录布局，下一步就是弄清楚如何将数据导入这些分区。

4.1.3 分区

分区是获取数据集并将其拆分为不同部分的过程。这些部分是分区，代表了对数据的有意义划分。数据中的公共分区示例是时间，因为它允许查询数据的人在特定的时间窗口内缩小范围。4.1.2节将时间作为决定在HDFS中布局数据的关键因素。

如果在HDFS中有一个大型数据集，需要对其进行分区，应该怎么做呢?本节将介绍两种可用于对数据进行分区的方法。

使用MultipleOutputs对数据进行分区

想象一下，你将股票价格数据流入HDFS，并且希望编写MapReduce作业，以根据当天的股票报价对数据进行分区。为此，你需要在单个任务中写入多个输出文件，让我们来看看如何实现这一目标。

问题

需要对数据进行分区，但大多数输出格式仅为每个任务创建一个输出文件。

解决方案

使用与MapReduce捆绑在一起的MultipleOutputs类。

讨论

Hadoop中的MultipleOutputs类绕过了在Hadoop中生成输出的正常通道。它提供了一个单独的API来写入分区输出，并将输出直接写入HDFS中的任务尝试目录，这可以继续提供给作业的Context对象的标准write方法来收集输出，还可以使用MultipleOutputs来编写分区输出。当然，你也可以选择仅使用MultipleOutputs类并忽略标准的基于上下文的输出。

在此技术中，你将使用MultipleOutputs按报价日期对股票进行分区。第一步是设置MultipleOutputs以便在工作中使用。 在驱动程序中，你将指明输出格式以及键和值类型：

为什么需要在驱动程序中命名输出?你可能想知道为什么MultipleOutputs要求指定输出名称(前面示例中的分区)。这是因为MultipleOutputs支持两种操作模式 - 静态分区和动态分区。

如果提前知道分区名称，静态分区可以正常工作，这为每个分区指定不同输出的格式提供了额外的灵活性(只需要对具有不同命名输出的MultipleOutputs.addNamedOutput进行多次调用)。对于静态分区，在调用addNamedOutput时指定的输出名称与在mapper或reducer中发出输出时使用的名称相同。

命名输出主要针对于动态分区，因为在大多数情况下，操作者不会提前知道分区名称。在这种情况下，仍然需要提供输出名称，当然这也可能被忽略，因为可以在mapper或reducer中动态指定分区名称。

正如以下代码所示，map(或reduce)类将获取MultipleOutputs实例的句柄，然后使用write方法写入分区输出。请注意，第三个参数是分区名称，即股票日期：

不要忘记close（）方法! 在任务完成后调用MultipleOutputs上的close方法非常重要。否则，输出可能会丢失数据，甚至造成文件损坏。

正如你在以下输出中所看到的，运行上一个示例会为单个mapper生成许多分区文件。我们还可以看到原始map输出文件，该文件为空，因为没有使用Context对象发出任何记录：

此示例使用了仅map作业，但在生产中，你可能希望限制创建分区的任务数。有两种方法可以做到这一点：

• 使用CombineFileInputFormat或自定义输入格式来限制作业中的mapper数量。

• 使用reducer明确指定合理数量的reducer。

总结

MultipleOutputs有很多值得关注的东西：它支持“旧”和“新”MapReduce API，并支持多种输出格式类。但是，使用MultipleOutputs应该注意一些问题：

• 在mapper中使用MultipleOutput时请记住，最终会得到NumberOfMappers * NumberOfPartition输出文件，根据经验，这会降低具有大量两个值的集群!

• 每个分区在任务期间都会产生HDFS文件句柄开销。

• 你经常会遇到大量小文件，这些文件会在分区程序的多次使用中累积。当然，这可以采用压缩策略来缓解(有关详细信息，请参阅第4.1.4节)。

• 尽管Avro附带了AvroMultipleOutputs类，但由于代码效率低下，速度很慢。

除了MultipleOutputs方法之外，Hadoop还附带了一个MultipleOutputFormat类，其功能类似于MultipleOutputs，主要缺陷是只支持旧的MapReduce API，并且只有一种输出格式可用于所有分区。

我们可以使用的另一种分区策略是MapReduce分区程序，它可以帮助减轻可能使用MultipleOutputs生成的大量文件。

使用自定义MapReduce分区程序

另一种分区方法是使用MapReduce中内置的分区工具。默认情况下，MapReduce使用分区程序来计算每个map输出键的散列，并对reducer的数量执行计数以确定应将记录发送到哪个reducer。我们可以编写自定义分区程序，然后根据分区方案路由记录来控制分区生成方式。

这种技术比以前的技术有额外的好处，因为通常会得到更少的输出文件，每个reducer只会创建一个输出文件，而MultipleOutputs则每个map或reduce任务都会产生N个输出文件，每个分区一个。

问题

对输入数据进行分区。

解决方案

编写自定义分区程序，将记录分区到适当的reducer。

讨论

自定义分区器可向MapReduce驱动程序公开辅助方法，允许定义从日期到分区的map，并将此map写入作业配置。然后，当MapReduce加载分区器时，MapReduce调用setConf方法; 在分区器中，我们将读取mapping到map中，随后在分区时使用。

驱动程序代码需要设置自定义分区程序配置。此示例中的分区是日期，如果希望确保每个reducer对应唯一的日期。股票示例数据有10个唯一日期，因此可以使用10个reducer配置作业，还可以调用先前定义的分区帮助程序函数设置将每个唯一日期map到唯一reducer配置。

除了从输入数据中提取股票日期并将其作为输出key之外，mapper几乎没有其他功能：

运行前面示例的命令如下：

此作业将生成10个输出文件，每个文件包含当天的股票信息。

总结

使用MapReduce框架自然地对数据进行分区可以带来以下几个优点：

• 对分区中的数据进行排序，因为shuffle将确保对流式传输到reducer的所有数据进行排序，这允许对数据使用优化的连接策略。

• 可以对reducer中的数据进行重复数据删除，这也是shuffle阶段的一个好处。

使用这种技术需要注意的主要问题是数据偏差，如果希望确保尽可能地将负载分散到reducer上，数据会存在自然偏差，这可能是一个问题。例如，如果分区是天数，那么大多数记录可能是一天，而可能只有一些记录可用于前一天或后一天。 在这种情况下，理想情况是将大多数Reducer分配到一天对记录进行分区，然后可能在前一天或后一天分配一两个记录，也可以对输入进行采样，并根据样本数据动态确定最佳reducer数量。

生成分区输出后，下一个挑战是处理分区产生的大量潜在小文件。

4.1.4 数据压缩

在HDFS中有小文件是无法避免的，也许你正在使用类似于前面描述的分区技术，或者数据可能以小文件大小有机地落在HDFS中。 无论哪种方式，都将暴露HDFS和MapReduce的一些弱点，比如：

• Hadoop的NameNode将所有HDFS元数据保留在内存中，以实现快速元数据操作。雅虎估计每个文件平均占用内存600字节的空间，转换为10亿个文件的元数据开销，总计60 GB，所有文件都需要存储在NameNode内存中。即使是当今的中端服务器RAM容量，这也代表着单个进程的大量内存。

• 如果对MapReduce作业的输入是大量文件，则将运行的mapper数量(假设文件是文本或可拆分的)将等于这些文件占用的块数。如果运行一个输入数千或数百万个文件的MapReduce作业，那么将花费更多时间在内核层处理创建和销毁map任务流程，而不是实际工作。

• 最后，如果在具有调度程序的受控环境中运行，则可能会限制MapReduce作业可以使用的任务数。由于每个文件(默认情况下)至少会生成一个map任务，因此这可能会导致作业被调度程序拒绝。

如果认为不会有这个问题，那就再想一想，文件百分比小于HDFS块大小吗?小多少?50%?70%?还是90%?如果大数据项目突然需要扩展以处理大几个数量级的数据集，该怎么办?这不是你首先使用Hadoop的原因吗?扩展就要添加更多节点，企业肯定不接受重新设计Hadoop并处理迁移文件。因此，在设计阶段最好尽早思考和准备这些可能的问题。

本节介绍了一些可用于在HDFS中组合数据的技术。首先讨论一个名为filecrush的实用程序，它可以将小文件压缩在一起以创建较少数量的较大文件。

使用filecrush压缩数据

压缩是将小文件组合在一起生成更大文件的行为，这有助于缓解NameNode上的堆压力。

就与Hadoop版本的兼容性而言，filecrush实用程序仅适用于Hadoop版本1。但是compacter(https://github.com/alexholmes/hdfscompact)以及Archive与Hadoop 版本2兼容。

问题

希望压缩小文件以减少NameNode需要保留在内存中的元数据

解决方案

使用filecrush实用程序。

讨论

filecrush实用程序组合或压缩多个小文件以形成更大的文件。该实用程序非常复杂，并能够

• 确定压缩文件的大小阈值(并通过关联，保留足够大的文件)

• 指定压缩文件的最大大小

• 使用不同的输入和输出格式以及不同的输入和输出压缩编解码器(用于移动到不同的文件格式或压缩编解码器)

• 使用较新的压缩文件交换较小的文件

我们在一个简单示例中使用filecrush ，粉碎一个小文本文件的目录，并用gzipped SequenceFiles替换。

首先，在HDFS目录人为创建10个输入文件：

运行filecrush，此示例使用新的大文件替换小文件，并将文本文件转换为压缩的SequenceFile：

运行filecrush后会发现输入目录中的文件已被单个SequenceFile替换：

还可以运行文本Hadoop命令来查看SequenceFile的文本表示：

你会注意到原始的小文件已全部移动到命令中指定的输出目录：

如果在没有--clone选项的情况下运行filecrush，则输入文件将保持不变，并且已压碎的文件将被写入输出目录。

输入和输出文件大小阈值

filecrush如何确定文件是否需要粉碎?通过查看输入目录中的每个文件，并将其与块大小进行比较(在Hadoop 2中，可以在-Ddfs.block.size命令中指定大小)。如果文件小于块大小的75%，则会被压缩。可以通过--threshold参数自定义此阈值，例如，如果需要将值提高到85%，则指定--threshold 0.85。

同样，filecrush使用块大小来确定输出文件大小。默认情况下，它不会创建占用超过八个块的输出文件，但可以使用--max-file-blocks参数进行自定义。

总结

Filecrush是一种将小文件组合在一起的简单快捷的方法。只要存在关联的输入格式和输出格式类，就支持任何类型的输入或输出文件。遗憾的是，它不能与Hadoop 2一起使用，并且在过去几年中项目中没有太多变化，因此可能企业不会考虑该程序。

因为filecrush需要输入和输出格式，所以如果正在处理二进制数据并且需要一种将小二进制文件组合在一起的方法，那么它显然还有不足之处。

使用Avro存储多个小型二进制文件

假设正在开发一个类似于Google图像的项目，可以在其中抓取网页并从网站下载图像文件，该项目是互联网规模，因此下载了数百万个文件并将它们单独存储在HDFS中。你已经知道HDFS不能很好地处理大量小文件，因此我们尝试一下Avro。

问题

希望在HDFS中存储大量二进制文件，并且无需遵守NameNode内存限制即可。

解决方案

在HDFS中处理小型二进制文件的最简单方法是将其打包到一个更大的文件中。此技术将读取存储在本地磁盘目录中的所有文件，并将它们保存在HDFS的单个Avro文件中。你还将了解如何使用MapReduce中的Avro文件来处理原始文件内容。

讨论

图4.2显示了该技术的第一部分，可以创建HDFS中的Avro文件。这样做可以在HDFS中创建更少的文件，这意味着存储在NameNode内存中的数据更少，这也意味着可以存储更多内容。

图4.2 在Avro中存储小文件以允许存储更多文件

Avro是由Hadoop的创建者Doug Cutting发明的数据序列化和RPC库。Avro具有强大的架构演进功能，其优势已经超过竞争对手，比如SequenceFile等，第3章详细介绍了这一内容，此处不再赘述。

请查看以下Java代码，该代码将创建Avro文件：

代码4.1读取包含小文件的目录，并在HDFS中生成单个Avro文件

要运行本章代码需要在主机上安装Snappy和LZOP压缩编解码器。有关如何安装和配置它们的详细信息，此处不作赘述。

当针对Hadoop的config目录运行此脚本时会发生什么(将$ HADOOP_CONF_DIR替换为包含Hadoop配置文件的目录)：

让我们确保输出文件是HDFS：

为确保一切正常运行，还可以编写代码以从HDFS读取Avro文件，并为每个文件的内容输出MD5哈希值：

此代码比写入更简单。由于Avro将架构写入每个Avro文件，因此无需在反序列化过程中向Avro提供有关架构的信息：

此时，我们已经在HDFS中拥有了Avro文件。尽管本章是关于HDFS的，但可能要做的下一件事是处理在MapReduce中编写的文件。这需要编写一个Map-only MapReduce作业，它可以读取Avro记录作为输入，并输出一个包含文件名和文件内容的MD5哈希值的文本文件，如图4.3所示。

图4.3 map作业读取Avro文件并输出文本文件

以下显示了此MapReduce作业的代码：

代码4.2 一个MapReduce作业，包含小文件的Avro文件作为输入

如果通过之前创建的Avro文件运行此MapReduce作业，则作业日志文件将包含文件名和哈希：

该技术假设使用的文件格式(例如图像文件)无法将单独的文件连接在一起。如果文件可以连接，应该考虑该选项。采用该方法请尽量确保文件大小与HDFS块一样，以最大限度减少存储在NameNode中的数据。

总结

我们可以使用Hadoop的SequenceFile作为保存小文件的机制。SequenceFile是一种比较成熟的技术，比Avro文件更成熟。但是SequenceFiles是特定于Java的，并且不提供Avro带来的丰富的互操作性和版本控制语义。

谷歌的Protocol Buffers以及Apache Thrift(源自Facebook)也可用于存储小文件。 但是都没有适用于本机Thrift或Protocol Buffers文件的输入格式。我们也可以使用另一种方法就是将文件写入zip文件。缺点是必须编写自定义输入格式来处理zip文件，其次是zip文件不可拆分(与Avro文件和SequenceFiles相反)。这可以通过生成多个zip文件并尝试使它们接近HDFS块大小来减轻。Hadoop还有一个CombineFileInputFormat，可以将多个输入拆分(跨多个文件)提供给单个map任务，这大大减少了运行所需的map任务数量。

我们还可以创建一个包含所有文件的tarball文件，生成一个单独的文本文件，其中包含HDFS中tarball文件的位置。此文本文件将作为MapReduce作业的输入提供，mapper将直接打开tarball。但是，这种方法会绕过MapReduce的局部性，因为mapper将被安排在包含文本文件的节点上执行，因此可能需要从远程HDFS节点读取tarball块，从而导致不必要的网络I/O。

Hadoop归档文件(HAR)是专门为解决小文件问题而创建的，是位于HDFS之上的虚拟文件系统。HAR文件的缺点是无法针对MapReduce中的本地磁盘访问进行优化，并且无法进行压缩。Hadoop版本2支持HDFS Federation，其中HDFS被划分为多个不同的命名空间，每个命名空间由一个单独的NameNode独立管理。实际上，这意味着将块信息保存在内存中的总体影响可以分布在多个NameNode上，从而支持更多数量的小文件。

最后，提供Hadoop发行版的MapR拥有自己的分布式文件系统，支持大量小文件。将MapR用于分布式存储对系统来说需要很大改变，因此，企业不太可能转移到MapR来缓解HDFS的这个问题。你可能会遇到想要在Hadoop中处理小文件的时间，并且直接使用它们会导致膨胀的NameNode内存和运行缓慢的MapReduce作业。此技术通过将小文件打包到更大的容器文件中来帮助缓解这些问题。我之所以选择Avro是因为它支持可拆分文件，压缩及其富有表现力的架构语言，这将有助于版本控制。如果文件很大，想要更有效地存储应该怎么办呢?这将在第4.2节得到解决。

4.1.5原子数据移动

分区和压缩等行为往往遵循类似的模式，比如在暂存目录中生成输出文件，然后需要在成功暂存所有输出文件后以原子方式移动到最终目标目录。这可能会带来一些问题：

• 使用什么触发器来确定已准备好执行原子移动?

• 如何在HDFS中原子移动数据?

• 数据移动对最终数据读者有何影响?

使用MapReduce驱动程序作为后处理步骤执行原子移动可能很好，但如果客户端进程在MapReduce应用程序完成之前死亡会发生什么?这是在Hadoop中使用OutputCommitter非常有用的地方，因为可以将任何原子文件移动作为工作的一部分，而不是使用驱动程序。

接下来的问题是如何在HDFS中原子移动数据。在最长的时间内，人们认为DistributedFileSystem类(这是支持HDFS的具体实现)上的重命名方法是原子的。但事实证明，在某些情况下，这不是原子操作。这在HADOOP-6240中得到了解决，但出于向后兼容性原因，重命名方法未更新。因此，重命名方法仍然不是真正的原子。相反，你需要使用新的API。如下所示，代码很麻烦，只适用于较新版本的Hadoop：

HDFS缺少的是原子交换目录能力，这在压缩等情况下非常有用。因为在这些情况下，你需要替换其他进程(如Hive)正在使用的目录的全部内容。有一个名为““Atomic Directory Swapping Operation”的项目可能有对你有些帮助。

以上是“怎么在HDFS中组织和使用数据”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注亿速云行业资讯频道！