Gobblin：统一Hadoop数据抽取

转载自（阿猫阿狗Hakuna）

摘要

      数据抽取是手机和分析大量数据的公司和组织的重要组成部分。本文描述了Gobblin，它是Hadoop的通用数据抽取框架，也是Linkedln最新的开源产品之一。在Linkedln，我们需要从各种来源抽取数据，比如关系存储、NoSQL存储、流系统、文件系统等等，抽取到我们的Hadoop集群。维护每个源的独立pipeline会导致各种操作问题。Gobblin旨在提供一个集中式的数据抽取框架来解决这个问题，该框架使得数据抽取变得更为简单。

      Gobblin与类似框架的核心区别在以下三个方面：通用性、可扩展性和可操作性。Gobblin支持开箱即用的数据源，并且可以轻松扩展以获得更多数据源支持。这使得组织能够使用单个框架来解决不同的数据抽取需求，从而使得操作变得简单。此外，通过端到端metrics和reporting模块，Gobblin可以简化和高效识别生产中的问题。

1.介绍

      大数据系统通常指它处理的数据集的数量，以及与之相关的强大处理能力和新的处理范式。然而，大数据系统复杂性的另一个重要方面来自于异构sources和sinks的共存。现实中，在开发人员和数据工程师进行传统ETL或数据处理之前，数据集成常常会成为很大问题。高效、准确的数据抽取与整合显得越来越重要。基于大数据抽取和集成痛点的亲身体验，我们构建了统一的数据抽取框架Gobblin。Gobblin的发展主要是因为Linkedln的数据源变得越来越异构。其中就包含了Expresso、Kafka、Oracle、MySQL、RocksDB等等。这些数据源每天收集TB级的数据，其中大部分需要加载到Hadoop集群中，以满足面向业务或面向用户的分析。我们曾经为每个数据源开发一个单独的数据pipeline，且曾经运行着十几个不同的pipeline。

      这些管道是由不同的团队开发的。不难想象这种方法的不可伸缩性，以及它在维护、可操作性和数据质量方面带来的问题。Gobblin的目标是将这些pipeline全部替换为通用的数据抽取框架，该框架易于配置，可以从多个不同数据源抽取数据，并且易于扩展，以适应新的数据源和使用场景。

      Gobblin面临的挑战有以下五方面：

（1）数据源整合：该框架为我们所有常用数据源（如MySQL、Kafka等）提供了开箱即用的适配器。

（2）处理模式：Gobblin支持standalone和可伸缩平台，包括Hadoop和Yarn。与Yarn的集成提供了持续抽取的能力。

（3）可扩展性：开发人员可以将自己的适配器集成到框架中，并使其可以用于社区中其他开发人员。

（4）Self-server：开发人员可以用自己的方式组合数据抽取和转换流，使用standalone模式在本地进行测试，并使用扩展模式在生产环境中部署，无需更改代码。

（5）数据质量保证：框架支持数据metrics收集器以及data quality checker（数据质量检查器），可以使用它们进行连续数据验证。

2.架构和概述

2.1 系统架构和组件

      图一展示了Gobblin的架构。一个Gobblin的job将数据从source（例如Kafka）中的数据抽取到sink（例如HDFS）中。一个job可能由多个workunits或task组成，每个workunits或task代表一个要完成的工作单元，例如，从Espresso表分区或Kafka主题分区抽取数据。

Gobblin：统一Hadoop数据抽取

2.1.1 Job Constructs（Job组件）

      一个job由很多constructs中的组件表示。Gobblin为每个constructs提供一个接口，可以通过实现这些接口来扩展Gobblin。

（1）Source（数据源）：Source负责将数据抽取工作划分为一组workunits，并为每个workunits指定Extractor（数据提取器）。这类似于Hadoop的InputFormat，它将输入分割为多个部分，并为每个部分指定一个RecordReader。生成workunits的算法通常尽可能平均地将工作分配给workunits。在我们的实现中，主要使用基于Hash的算法。

（2）Extractor（抽取器）：抽取器执行workunits中指定的工作，即抽取指定的数据记录。workunits应该包含数据的源信息（比如，从Kafka的哪个分区，哪个DB表等等），同时需要包含抽取数据的哪个部分。Gobblin使用watermark对象告诉Extractor开始记录（low watermark）和结束记录（high watermark）是什么。例如，对于Kafka作业，watermark可以是分区的offsets，对于DB，watermark可以是列的值。watermark可以是任意类型的，不一定非要是数字，只要Extractor能够知道从哪里开始，哪里结束。

（3）Converter（转换器）：Converter在抽取数据时动态地进行数据转换。通常一些场景需要这种转换，比如，schema中的一些字段需要被隐藏起来；或者从Source抽取的数据是字节数组或JSON对象，需要转换为所需的输出格式。Converter可以将一条输入记录转换为0或多条记录。Converters是可插拔的和可链接的，实现很简单，只需要实现两个方法，分别是转换模式以及数据记录。

（4）Quality Checker（数据质量检查器）：Quality Checker确定抽取的记录是否良好并可以发布。就检查范围而言，存在两种检查策略：record-level策略和task-level策略。他们分别检查单个记录的完整性和整个任务的输出。根据需要，还有两种类型的质量检查策略：强制策略和可选策略。违反强制策略会导致记录或任务输出被丢弃或写到异常值文件夹，违反可选策略会导致一个警告。

（5）Writer（写入器）：Writer将任务提出的记录写入适当的sinks。它首先将记录写入一个临时目录（staging directory）。任务成功完成并将所有记录写入staging目录后，Writer将它们移动到输出目录，该目录根据质量检查策略进行检查，并由Publisher（发布器）发布。Gobblin的数据写入器可以将数据写到HDFS，Kafka，S3等等，或者不同的格式（比如CSV,Parquet，Avro等）。

（6）Publisher（发布器）：发布器将Writer写入的数据发布到最终job输出目录（以原子方式）。Gobblin提供两个提交策略：commit-on-full-succes，只在所有任务成功时提交数据；commit-on-partial-success，只提交成功部分数据。

      除了这些constructs外，Gobblin还支持通过fork操作符在任务流中进行分支操作，允许将记录交由不同的分支处理。每个fork都有自己的converters，quality checker，writers以及publishers，不同的fork可能会将记录发布到不同的sink，或者以不同格式发布到相同的sink。

      一个典型的作业构造流程如图2所示：

Gobblin：统一Hadoop数据抽取

2.1.2 State Store（状态存储）

      Gobblin使用元数据（也称为State Store）管理job和task状态。每次作业运行完成后，其high watermark和其他运行时的元数据将被持久化到State Store中。同一个任务下一次运行将从State Store中获取上一次运行的元数据，以便可以从上一个运行时的位置接着运行。

      Gobblin的State Store有一个默认实现，它将job和task的状态序列化到Hadoop SequenceFiles中。每个job都有一个单独的目录，其中存储其运行的state store SequenceFiles。开始时，作业每次运行都会读取对应目录中上一次运行保存的SequenceFile，以获取其运行时元数据。为了防止State Store无限制增长，Gobblin提供了一个工具去清理State Store，基于可配置的保留设置。Gobblin还允许用户通过配置属性插入自己的State Store实现，只要符合State Store接口定义即可。

2.1.3 Job Execution（Job执行）

      job被创建后，Gobblin job execution runtime负责在指定部署模式的平台上运行作业。常见的比如job/task调度、job/task状态管理，错误处理和重试、监视和报告等都由runtime处理。

      对于错误处理，Gobblin针对job和task失败采用了多级预防。对于job失败，Gobblin会连续跟踪其失败次数，如果超过了定义的阈值，可以发送电子邮件，以便job的所有者能查看失败情况。对于task失败，Gobblin可以配置最多重试次数。

        对于作业调度，Gobblin可以集成Oozie，Azkaban，或Chronos，Gobblin内置的调度器是Quartz，用在standalone部署模式中。Gobblin将作业调度器和作业解耦，以便不同作业可以在不同位置运行。

2.1.4 Metrics and Monitoring（度量和监控）

      Gobblin提供了用于metric收集和报告的端到端系统，使用通用的metric库进行监视。该库支持各种类型的度量，包括计数器、直方图、仪表灯，且能够将报告发送给多个后端系统，比如Kafka，Graphite，JMX，以及日志。

2.1.5 Compaction（压缩）

      Gobblin中有一个压缩模块用于数据压缩。它提供了两个开箱即用的压缩器，一个使用Hive，另一个使用MapReduce，两者都执行重复数据删除。例如，在Linkedln，我们将时间数据放入按照小时分隔的文件夹中，每天压缩它们一次进入到按天分隔的文件夹中，并删除重复内容。

2.1.6 Deployment（部署）

      Gobblin能够在不同平台上以多种部署模式运行，以满足不同的伸缩性和资源使用需求。目前，Gobblin可以使用专用线程池在一台机器上以standalone模式运行任务，也可以在Hadoop上以MapReduce模式运行任务，使用mappers来运行任务。这使得我们能够在standalone模式下测试一个flow，并使用MapReduce模式将其部署到生产环境中，而无需更改代码。最新支持的第三个部署选项是在Yarn上运行Gobblin，以实现更灵活和更高效的资源管理，以及作为长期服务运行的能力。这对于从Kafka这样的流数据源抽取数据非常方便。Gobblin on Yarn是在Apache Helix和Apache ZooKeeper的基础上设计和构建的。

      具体的说，它使用Helix管理在Yarn container中运行的workunits，ZooKeeper管理Yarn ApplicationMaster和containers的资源协调和消息传递。ZooKeeper还被用作运行的workunits元数据的中央存储库。Helix管理集群中分布式资源，它保证了workunits在可用的containers间均匀分布，且负责在container失败或关闭时进行rebalance。Gobblin on Yarn的一个重要特性是有助于实现更有效的资源管理，可以适应不断变化的工作负载，并在运行时动态调整Yarn Container集群的大小。

2.2 生产环境中的Gobblin

      Gobblin已经在Linkedln投入生产6个多月，目前负责从12个不同的数据源抽取超过300个数据集。这些数据源有几种不同类型，包括MySQL、SQL Server、SFTP Server、S3以及REST端点。添加一个新的数据集很简单：工程师只需要提供一个简单的配置文件，就可以将数据导入HDFS。我们的ETL和解决方案工程师不再需要数据抽取的问题，可以只关注于数据的使用。

3.案例学习

      在本节中，我们将详细介绍如何使用Gobblin从Linkedln上的Kafka和JDBC数据源抽取数据，通过这些例子来解释如何为不同的需求扩展Gobblin。

3.1 Kafka到HDFS的抽取

      在Linkedln，我们抽取超过3000个主题，成千上万个分区的数据不断地从Kafka到Hadoop集群。这些主题包括用户活动（例如页面查看和发送电子邮件），metrics，以及需要跟踪的内部事件。Kafka抽取目前由Camus完成，这是一个专门为kafka-HDFS设计的开源pipeline。为了更好的性能、稳定性、可操作性以及数据完整性，Gobblin的Kafka适配器正在取代Camus。

      Gobblin目前以小批量、连续的方式抽取Kafka记录。如果使用Yarn部署模式，则能够以长时间运行的流模式运行Kafka抽取。接下来，我们将详细介绍Kafka适配器中的每个构造是如何设计和实现的。

      Source：回想一下，Souce负责生成workunits。要从Kafka抽取数据，需要实现一个生成Kafka workunits的Source。Kafka的workunits应该为每个分区包含开始和结束的offsets（即，low watermark和high watermark）。

      对于应该抽取的每个主题分区，我们首先检索上一个运行所拉取的最后一个offset，它应该是当前运行的开始offset。我们还从Kafka集群中检索当前可用的最早和最晚的offsets，并验证起始offset位于最早和最晚offsets之间。最新可用offset是当前workunits要拉取的结束偏移量。由于新记录会不断发布到Kafka，旧记录会根据保留设置被删除，所以分区的最早和最新偏移量会不断变化。这增加了数据抽取pipeline的压力，因为它的速度必须比发布到Kafka的记录速度快。

      对于每个分区，在确定开始和结束offsets之后，将创建一个workunit。在得到所有初始workunits后（每个partition一个），使用基于bin-packing的算法执行两个步骤：（1）将于相同topic的partition对应的一些workunits合并到更大的workunits中，以便这些partition可以写入相同的输出文件，从而减少在HDFS上发布的小文件的数量，从而减少NameNode上的压力。（2）为containers分配workunits并平衡containers的工作负载。

      Extractor：如上所述，一个Kafka的workunit具有相同主题的一个或多个partition。为每个workunit创建Extractor，如果无法提取partition，则跳过（并在metrics中报告失败），但不是使该task失败，这样的话不同的partitions是隔离的，不会互相影响。

      抽取所有partition后，Extractor将每个partition的最后一个offset报告给状态对象，该对象会被持久化到State Store。

      Converter：Linkedln的Kafka集群主要存储Avro格式的记录。最常用的是删除字段的converter，因为某些敏感字段需要从Avro记录中删除才能发布。

      Quality Check：由于我们要将Kafka数据放入时间分区（小时和天计），我们需要使用一个强制的record-level记录质量策略来检查每条记录是否有一个时间戳字段。如果是，可以将其写入适当的文件夹；否则必须将其丢弃或写入异常值文件夹。我们还使用另一个策略来验证记录的主键字段是否存在且是否为非空。

      Writer and Publisher：对于Kafka的抽取，我们实现了按时间分区的Writer和Publisher，它们根据时间戳将每条记录写入并发布到适当的时间分区文件夹。

3.2 JDBC到HDFS的抽取

      JDBC协议为使用Java代码查询和操作各种关系型数据库提供了一个清晰的抽象。今天，Linkedln从MySQL和SQL Server中抽取大约250个表。

      JDBC抽取和Kafka抽取之间的一个重要区别是数据库元组是可变的，而Kafka记录不是。因此，JDBC抽取不仅需要抽取新的元组，还需要一种处理更新元组的方法。在本小节讨论JDBC抽取的constructs之后，我们将解释如何做到这一点。

      Source：Gobblin为JDBC提供了两个源：一个MySQL的Source和一个SQL Server的Source。它们使用相同的算法生成workunits，唯一的区别是MySQL源使用MySQL Extractor，SQL Server源使用SQL Server Extractor。

      要为应该抽取的每个表生成workunits，我们要检测上次运行提取的元组的最新时间戳，并将其作为low watermark。当前时间作为high watermark。然后我们将从low watermark和high watermark间的间隔分隔成相等长度的可配置分区数，为每个分区创建一个workunits。

      Extractor：MySQL和SQL Server的Extractor使用SQL查询从数据库中获取具有指定时间戳的元组，并将它们作为JSON返回。与Kafka Extractor类似，它会报告抽取的最新时间戳，并将其持久化到State Store中，用于下一次运行。

      Converter and Quality Checker：我们以Avro格式将抽取的数据存储在HDFS中。因此，我们使用Converter将JSON元素转换为Avro记录。我们还使用一个task-level的质量检测策略确保写入的记录数量与count查询返回的行数相同。

      Writer and Publisher：我们使用一个简单的Writer和Publisher，将每个表的记录发布到一个文件夹中。

      handling Tuple Updates（处理tuple更新）：如上所述，在JDBC抽取中，我们需要一种处理元组更新的方法。一种简单的方法是在每次运行时全部拉出每个表，这对于大型表来说几乎不可行，对于两次运行之间只有一小部分更新元组的表来说效率很低。

      对于这些表，我们使用了一种改进的方法，其中涉及用于JDBC抽取的多个job类型。（1）snapshot job(提取整个表)的调度频率相对较低；（2）append job只提取修改时间戳晚于上次运行(可能是snapshot job，也可能是append job)提取的最新时间戳的元组。append job以更高的频率调度。注意，这要求表具有一个修改时间戳列。还要注意，提取的snapshot和append数据文件中可能存在元组的不同版本。

      如果用户的应用可以接受一个tuple的多个版本，或者如果存在多个版本，用户可以选择tuple的最新版本，那么用户可以直接使用抽取的snapshot和append文件。除此之外，Gobblin在2.1.5节中引入了一个压缩模块，可用于压缩snapshot和append文件。

      Apache Sqoop是另一个数据抽取框架，它的焦点是SQL到Hadoop数据抽取。与Sqoop相比，在Gobblin中，我们的目标是提供合适粒度的操作符，这些操作符可以很容易地组合到抽取流中，并且可以很容易地针对不同的使用场景进行扩展。Converter、Quality Check和fork操作符是此类操作符的几个示例，目前Sqoop中还没有这些操作符。此外，正如解释的那样，Gobblin支持多种类型的抽取和压缩。虽然Sqoop支持增量抽取，但它没有类似于snapshot-ppend作业的功能，也不支持压缩。

4.相关工作

      与Gobblin最具可比性的系统是Apache Sqoop，它最初被设计为SQL-Hadoop抽取工具，后来成为多种类型Source和Sink的通用框架。接下来，我们将在几个类别中总结Gobblin和Sqoop之间的主要区别:

（1）在目标方面，Gobblin从一开始的目标之一就是将连续流抽取与批数据抽取统一起来。我们认为，这两种抽取模式不需要完全不同的、专用的引擎，相反，它们可以由一个具有适当抽象级别的单一框架来处理，该框架对如何定义数据抽取工作单元进行了适当的抽象。这是与Sqoop最大的区别之一，Sqoop的架构是面向批处理的抽取解决方案。

（2）在模块化和组件化方面，Sqoop没有提供Gobblin中的一些模块作为数据抽取作业的构建块，如Converter、Quality Checker和fork操作符，如3.2节所述。

（3）在数据语义方面，Gobblin支持多种类型的抽取和压缩，如3.2节所述。Sqoop不支持这些。

（4）在job/task状态管理和故障处理方面，State Store在Gobblin内部维护，用于管理作业状态，如2.1.2节所述。Gobblin使用watermark来管理增量拉取，这种类型可以是简单类型，也可以是复杂类型。相比之下，Sqoop为增量导入提供了一个功能不太强大的特性，它基于单个属性上的单个数值。据我们所知，Sqoop也不提供维护此类检查点和其他作业状态的方法。

（5）在可操作性方面，Gobblin有一个完整的端到端度metrics集合和报告系统，在2.1.4节中进行了解释。