2.2 大数据采集技术

大数据采集可以细分为数据抽取、数据清洗、数据集成、数据转换等过程，将分散、零乱、不统一的数据整合到一起，以一种结构化、可分析的形态加载到数据仓库中，从而为后续的数据使用奠定坚实基础。

数据采集可以分为内部采集与外部采集两个方面。

内部数据采集技术主要包括：

（1）离线数据采集技术，首先要是基于文件的数据采集系统、日志收集系统等，代表性的工具有Facebook公司开发的Scribe、Cloudera公司开发的Flume和Apache基金会支持的Chukwa等；其次是基于数据库和表的数据采集技术，基于数据库的数据采集系统中代表性工具有GoldenGate公司的TMD、迪思杰公司而数据采集软件、IBM公司的CDC（InfoSphere Change Data Capture，CDC）、MySQL支持的Binlog采集工具等；在基于表的批量抽取软件中，广泛应用的是Sqoop和其他ETL工具。

（2）在线数据采集技术，主要是基于消息的采集、数据流采集等。基于消息采集的技术，如性能数据采集等，代表性的产品有Linkedin的Kafka，以及开源的ActiveMQ、RabbitMQ、RocketMQ等。基于数据流的采集技术，如信令数据采集等，代表性的产品有IBM StreamBase、Twitter公司的Storm等。这些工具或组件一般会根据场景选择压缩算法。

外部数据采集主要是指互联网数据的采集，相关技术主要分为两类。

（1）网络爬虫类，即按照一定的规则，自动抓取互联网信息的程序框架，例如，用于搜索引擎的网络爬虫属于通用网络爬虫，商用的代表性产品有Google、Baidu公司开发的系统等，其网络搜索技术已经非常成熟，但是并不对外开放技术。开源的技术有Apache Nutch、Scrapy、Heritrix、WebMagic、WebCollector等网络爬虫框架。

（2）开放API类，即数据源提供者开放的数据采集接口，可以用来获取限定的数据。在外部数据中，除了互联网数据采集技术，也有基于传感器应用的采集技术，这种技术在物联网中用得较多。此外，还有电信公司特有的探针技术，例如，我们在打电话、利用手机上网时，电信公司的路由器、交换机等设备中都会有数据交换，探针就是从这些设备上采集数据的技术。

目前，数据抽取、清洗、转换面临的挑战在于：数据源的多样性问题、数据的实时性问题、数据采集的可靠性问题、数据的杂乱性问题。这里要特别指出的是，通过采集系统得到的原始数据并不是干净的数据，大部分的数据都是带有重复、错误、缺失的所谓脏数据。实际上，数据科学家几乎80%的工作都是处理这些脏数据，可见由数据的杂乱性带来的麻烦是非常大的。因此，如何高效精准地处理好这些原始数据，也是大数据采集技术研究面临的重大挑战。

2.2.1 结构化数据采集工具

在Hadoop大数据应用生态系统中，Sqoop作为Apache的顶级项目，主要用来在Hadoop和关系数据库之间传递数据。通过Sqoop可以方便地将数据从关系数据库导入HDFS、HBase或Hive中，或者将数据从HDFS导出到关系数据库中。图2-2是Sqoop系统架构示意图。

Sqoop系统架构非常简单，主要通过JDBC和关系数据库进行交互。从理论上讲，支持JDBC的数据库都可以使用Sqoop和HDFS进行数据交互。

Sqoop系统数据具有以下特点：

• 支持文本文件、Avro Datafile、SequenceFile；
• 支持数据追加，可通过apend指定；
• 支持表的列选取，支持数据选取，可和表一起使用；
• 支持数据选取，如读入多表连接（join）后的数据，不可以和表同时使用；
• 支持Map数定制；
• 支持压缩；
• 支持将关系数据库中的数据导入Hive（Hive-import）、HBase（HBase-table）中。

2.2.2 日志收集工具与技术

日志收集是大数据的基石，企业内部的业务平台每天都会产生大量的日志数据，这些日志数据可供离线和在线的分析系统使用。高可用性、高可靠性和高扩展性是日志收集系统需要具有的基本特征。

1. 日志收集

日志收集模块需要使用一个分布式的、具有高可靠性和高可用性、能够处理海量日志数据的框架，并且应该能够支持多源采集和集中存储。目前常用的开源日志收集系统有Flume、Scribe等。Flume是由Cloudera开发的一个分布式、高可靠性和高可用性的海量日志收集系统，支持在系统中定制各类数据发送方，用于收集数据；同时，Flume也可对数据进行简单处理，并写入各种数据接收方（可定制）。

Flume的工作流程是先收集数据源的数据，再将数据发送到接收方。为了保证这个过程的可靠性，在发送到接收方之前，会先对数据进行缓存，等到数据真正到达接收方后，才会删除缓存的数据。

Flume传输数据的基本单位是事件（Event），如果是文本文件，则通常是一行记录，这也是事件的基本单位。事件（Event）从源（Source）传输到通道（Channel），再从通道传输到目的地（Sink），事件本身是一个字节数组，并可携带消息头（Headers）信息。

Flume运行的核心是Agent，Agent本身是一个Java进程，一般情况下Agent由三个组件构成，分别是Source、Channel和Sink。通过这三个组件就可以完成整个数据收集工作，使Event能够从一个地方流向另外一个地方。图2-3给出了Flume的工作流程。

Source：可以接收外部数据源发送的数据，不同的Source可以接收不同的数据格式。例如，Spooling Directory可以监视指定文件夹中文件的变化，如果该目录中有新文件产生，就会立刻读取该文件中的内容。

Channel：用来接收Source输出数据的缓存池，Channel中的数据在进入Sink并成功发送出去后或者进入终端时才会被删除，当Agent内部发生写入故障时不会造成数据的丢失，保证了数据收集的高可靠性。

Sink：用于接收Channel中的数据，发送给数据接收方或者其他Source，例如数据可以写入HDFS或者HBase中。

2. 数据分发工具Kafka

Flume收集的数据和进行日志处理的系统之间可能存在多对多的关系，为了解耦和保证数据的传输延迟，可以选用Kafka作为消息中间层进行日志中转分发。Flume发送源数据流的速度不太稳定，有时快有时慢，当Flume的数据流发送速度过快时（这种情况很常见），会导致下游的消费系统来不及处理，这样可能会丢弃一部分数据。Kafka在这两者之间可以扮演一个缓存的角色，而且数据是写入到磁盘上的，可保证在系统正常启动/关闭时不会丢失数据。

Kafka是Apache开发的一个开源分布式消息订阅系统，该系统的设计目标是给实时数据处理提供一个统一、高吞吐量、低等待的平台。Kafka提供了实时发布订阅的解决方案，克服了实时数据消费和更大数量级的数据量增长的问题，Kafka也支持Hadoop中的并行数据加载。图2-4是Kafka的架构图。

Kafka架构包含几个重要的组成部分，如Kafka集群的Broker、生产者（Producer）、消费者（Consumer）。Kafka在保存消息时会根据Topic进行归类，发送消息者称为Producer，消息接收者称为Consumer。此外，Kafka集群由多个Kafka实例组成，每个实例（Server）称为Broker。无论Kafka集群还是Producer和Consumer，都是通过ZooKeeper来保证系统可用性的。

（1）Topic：消息的基本单位。一个Topic可以看成一类消息，Kafka在保存消息时会按照Topic进行归类，每个Topic具体的数据存放位置由配置文件来决定，可能会存放到不同的分区（Partition）上。每条消息在文件中的位置称为偏移量（Offset），偏移量是一个数据类型为长整型（long）的数字。

（2）Broker：Kafka集群的基本单位，Kafka集群中包含多个Broker。Kafka集群中可能存在一个或多个代理（Broker）服务器，负责Producer和Consumer二者之间的消息处理与交互。

（3）Producer：生产者，生产（发布）Topic的进程。

（4）Consumer：消费者，消费（订阅）Topic的进程，同时若干个消费者还可以组成一个消费组，这样当生产者发布Topic时可以实现以下两种常用功能。

• 只针对某一些ID或某组ID对应的消费者通过点播发布消息；
• 通过广播将消息发给所有的消费者。

图2-5给出了Kafka的应用流程。

Kafka需要使用ZooKeeper（分布式应用管理框架）进行协调，从而保证系统的可用性，以及保存一些元数据（Meta Data）。ZooKeeper与Broker、Producer、Consumer之间是通过TCP协议进行通信的。

Kafka的典型使用场景如下。

（1）消息系统（Message System）。对于一些常规的消息系统，Kafka是个不错的选择。分区、多复本和容错等机制可以使Kafka具有良好的扩展性和性能优势。不过，到目前为止，Kafka还没有提供JMS中的事务性消息、传输担保（消息确认机制）、消息分组等企业级特性。Kafka只能作为常规的消息系统使用，并不能确保消息发送与接收绝对可靠。

（2）网站活性跟踪（Websit Activity Tracking）。Kafka作为网站活性跟踪的最佳工具时，可以将网页/用户操作等信息发送到Kafka中，并进行实时监视或者离线统计分析等，例如，各种形式的Web活动产生的大量数据，用户活动事件（如登录、访问页面、单击链接），社交网络活动（如喜欢、分享、评论），以及系统运行日志等，由于这些数据的高吞吐量（每秒百万级的消息），因此通常由日志收集系统和日志聚合系统来处理。这些传统方案可将日志数据传输给Hadoop来进行离线分析。但是，对于需要实时处理的系统，就需要其他工具的支持。

（3）日志聚合系统（Log Aggregation System）。Kafka的特性使它非常适合作为日志聚合系统，可以将操作日志批量、异步地发送到Kafka集群中，而不是保存在本地或者数据库中。Kafka可以批量地提交消息、压缩消息等，这对Producer而言，几乎感觉不到性能的开展。此时Consumer可以使Hadoop来进行存储和分析。

总之，Kafka是一个非常通用的系统，允许多个Producer和Consumer共享多个Topic。相比之下，Flume主要用于向HDFS、HBase发送数据，它对HDFS进行了特殊的优化，并且集成了Hadoop的安全特性。如果数据被多个系统消费，则建议使用Kafka；如果向Hadoop发送数据，则建议使用Flume。