第1篇：通过流式数据集成实现数据价值（1）
第2篇：通过流式数据集成实现数据价值(2)
本篇为通过流式数据集成实现数据价值的第3篇——实时持续数据收集。
作为所有流式数据集成解决方案的起点，需要实时持续收集数据。 这被称为“流优先”方法，如果没有此初始步骤，流式数据集成和流分析解决方案都无法执行。实现此方法的方式因数据源不同而不同，但都具有一些共同的要求：
- 由源生成数据后立即收集
- 从源中捕获元数据和模式信息，以与数据一起存放
- 将数据转换为通用事件结构以用于处理和交付
- 记录源位置
- 处理数据模式更改
- 通过多线程和并行扩展
- 通过恢复处理错误和故障场景，以确保不会丢失任何数据
以下各节说明了我们如何针对各种不同的源类别（数据库，文件和日志，消息传递系统，云和API，以及设备和IoT）实施这些要求，并将提供示例以阐明每种情况。
3.1 数据库和变更数据捕获（CDC）
数据库表示一些实际应用程序的当前状态，在事务处理上下文中最有意义。应用程序提交来自许多网络端点的查询和更新，这些端点作为一系列事务进行管理，以便进行状态观察和转换。
从20世纪70年代末到本世纪初，“数据库”一词通常被用来指关系数据库，其中基础实体和这些实体之间的关系被建模为表。在过去的二十年中，这个术语已经成为关系数据库系统和新兴的NoSQL系统的统称，而NoSQL系统又成为键值存储、文档存储、图形数据库等的统称。这些数据库可以是集中式的，也可以是分布式的。它们也可以在本地维护或存储在云中。
然而，由于数据库表示其中数据的当前状态，并且查询它们只返回该状态，因此它们并不天生适合通过查询机制进行流式数据集成。需要使用另一种方法将数据库转换为流数据源：CDC。
当应用程序与数据库交互时，它们使用插入、更新和删除操作数据。CDC直接拦截数据库活动，并收集发生的所有插入、更新和删除，将它们转换为流事件。
3.1.1 CDC 方法
有几种CDC方法已经使用了几十年，每种方法都有自己的优点，具体取决于用例。在时间敏感的高速数据环境中，低延迟、可靠和可伸缩的CDC支持的数据流是必需的。
关系数据库中捕获的业务事务对于理解业务操作的状态至关重要。传统的基于批处理的方法每天移动数据一次或多次，会带来延迟，并降低组织的操作价值。当新的数据库事件发生时，CDC通过不断地移动和处理数据来提供实时或接近实时的数据移动。全天不断地移动数据，也更有效地利用了网络带宽。
有以下几种CDC方法可以识别需要捕获和移动的更改。让我们来讨论一下每种CDC方法的优点和缺点:
- 时间戳 通过使用现有的LAST_UPDATED或DATE_MODIFIED列，或者在应用程序中添加一个不可用的列，您可以在应用程序级创建自己的CDC解决方案。此方法仅检索自上次提取数据以来已更改的行。此方法中的数据完整性可能存在问题;例如，如果删除了表中的一行，那么这一行将没有DATE_MODIFIED列，并且不会捕获删除操作。这种方法还需要CPU资源来扫描表中的更改数据和维护资源，以确保在所有源表上可靠地应用DATE_MODIFIED列。
- 表差异 通过比较将要在源系统和目标系统中复制的表(通过运行差异比较)，此方法仅加载不同的数据以支持一致性。虽然这对于管理已删除的行效果更好，但是识别差异所需的CPU资源非常重要，并且需求会随着数据量的增加而增加。差异比较方法还会带来延迟，无法实时执行。
- 触发器 在应用程序级别构建CDC的另一种方法是定义触发器并在影子表中创建您自己的更改日志。在插入、更新或删除命令(指示更改)之前或之后触发触发器，用于创建更改日志。在SQL级别操作时，有些用户更喜欢这种方法。但是，源数据库中的每个表都需要触发器，而且在进行更改时，与操作表上运行的触发器相关的开销更大。除了对应用程序的性能有重大影响之外，在应用程序更改时维护触发器还会导致管理负担。许多应用程序用户不希望在操作表中引入触发器，从而危及应用程序行为。
- 基于日志 数据库包含存储所有数据库事件的事务(有时称为重做)日志，以便在发生崩溃时恢复数据库。使用基于日志的CDC，可以从源数据库的事务或重做日志中读取新的数据库事务(包括插入、更新和删除)。捕获更改时不需要进行应用程序级的更改，也不需要扫描操作表，这两者都会增加额外的工作负载并降低源系统的性能。
3.1.2 基于日志的CDC最适合流式数据集成
CDC，特别是基于日志的CDC，在过去20年中变得流行起来，因为企业组织已经发现，共享来自在线事务处理(OLTP)数据库的实时事务数据可以实现各种各样的用例。快速采用云解决方案需要从内部数据库构建实时数据管道，以确保云系统持续更新。将企业数据库转换为流数据源，而不受批处理时间窗口的约束，为今天的现代数据架构奠定了基础。
出于多种原因，流集成应该利用基于日志的CDC。它最小化了源系统的开销，减少了性能下降的机会。此外，它是非侵入性的。它不需要对应用程序进行更改，比如向表中添加触发器。它是一种轻量级的，但也是一种获取更改数据的高性能方法。尽管从数据库日志中读取数据操作语言(DML)操作(插入、更新、删除)，但是这些系统仍然可以为最终用户提供高性能的运行。
通过CDC接收变更数据只是第一步，但也是最重要的一步。 此外，流式数据集成平台需要整合以下内容：
- 来自多个数据库的基于日志的CDC，用于非侵入性，低影响的实时数据提取，以最大程度地减少源上的CPU开销，并且不需要更改应用程序。
- 从多个并发数据源中提取数据，以将数据库事务与半结构化和非结构化数据结合在一起。
- 端到端变更数据集成，包括：
由于数据库系统中跟踪的数据的性质，下游应用程序不能容忍数据丢失。这意味着，如果外部数据库系统或CDC进程发生故障，事件检查点必须确保CDC进程仔细跟踪最早期的活动事件。
必须保留精确一次处理(E1P)和/或至少一次处理保证。这需要理解消费系统及其支持的原子性语义。
事务按提交顺序或生成顺序传播。因此，按事务顺序生成的数据必须能够保留该顺序，并在源端按需要保留事务边界。
当从数据库日志中获取更改数据时，提交的事务应该维护它们的事务上下文。在整个数据移动、处理和交付步骤中，应该保留这个事务上下文，以便用户可以创建可靠的副本数据库。
用户应该能够在移动数据时对其进行过滤，聚合，掩盖，转换和丰富其变化，而不会丢失交易环境。
修改源数据库架构并创建数据定义语言（DDL）语句后，流集成平台应能够在不暂停的情况下将模式更改应用于目标系统。
- 模式变更复制
- 移动中变更数据处理
- 事务完整性
- 事件顺序保证
- 活动交付保证
- 零数据丢失保证
- 将变更数据转换为对时间敏感的洞察。除了为变更数据构建实时集成解决方案之外，还应该能够对变更数据执行流分析，以获得即时的洞察。
基于日志的CDC是将数据库转换为流数据源的主流方法。然而，获取变更数据只是流式数据集成解决方案应该解决的许多问题中的第一个。
3.2 文件和日志
许多应用程序，如web服务器、应用服务器、物联网边缘服务器或企业应用程序，不断地生成写入文件或日志的数据记录。这些文件可以位于本地磁盘子系统、分布式文件系统或云存储中。
这些数据包含了运营分析所需要的有价值的信息。在批处理提取、转换和加载(ETL)系统中，这些文件在被ETL读取之前被写入并关闭。
但是，对于实时系统，必须能够对当前写入的文件(打开的文件)执行实时数据收集。
从文件系统收集数据
收集实时文件数据需要一套算法来检测文件/目录/节点的变化:
- 理解基础文件格式的内容，以便能够解析文件记录
- 维护位置偏移，以反映后续收集的当前EOF(文件结束)标记
- 识别破损/部分记录
- 针对各种故障场景的恢复处理
传统的ETL成功地在文件完成后从文件中提取数据。但是对于实时处理，需要在写入新记录时立即收集新记录，以使传播延迟的粒度低于文件大小。
在正在进行的文件生成过程中，实时流处理中出现了几个常见的模式，这些模式需要支持，并且会带来重大的技术挑战。一些例子包括:
- 支持多个文件系统，包括Linux (ext*)、Windows (NTFS)、Hadoop (HDFS)、基于网络(NFS)、云存储系统(AWS S3、Azure ADLS、谷歌GCS等)。
- 支持多种文件格式，如JSON、DSV、XML、Avro、Thrift、Protocol Buffers和Binary。
- 支持从需要读取文件的多个目录和子目录中读取。不可能总是有一个可以生成所有文件的中央存储库。
- 支持使用静态和动态记录分隔符进行数据解析。
- 支持在文件和目录级别使用通配符进行数据收集。
- 当文件按顺序排列并翻转到基本顺序时，支持数据收集。
- 管理打开文件描述符的数量。
- 关于数据丢失，至少一次或最多一次处理的事件保证。
- 处理架构更改。
3.3 消息息系统
在可以为流式数据集成提供数据的所有类型的源中，消息传递系统是最自然的选择。它们本质上是实时的，并将数据推送给消费者。实际上，消息传递系统通常是流集成解决方案的必需组件，这对于数据的连续移动是必需的。消息传递系统通常由将消息传递给代理以供消费者阅读的生产者组成。为了持续从消息传递系统收集数据，流式数据集成解决方案需要能够以消费者身份连接到代理。
在过去几年中，随着云技术的迅速普及，云提供商还引入了消息传递系统。Microsoft Azure Event Hub，Amazon Kinesis和Google Pub/Sub都提供了基于云的消息传递平台，旨在灵活地扩展和支持云中的流和消息驱动的应用程序。
由于异构集成和来自任何企业（或云系统）的数据收集是流式数据集成的重要部分，因此您需要考虑所有这些不同类型的消息传递系统。鉴于大多数此类系统每秒可处理数万至数百万条消息，因此连续收集的可伸缩性是关键。
3.3.1 从消息系统收集数据
使用消息传递系统时，有两个主要注意事项。首先，系统需要连接到消息传递提供程序并使用某种API订阅以接收消息。为此，通常需要在消息传递适配器内解决安全性，压缩、加密和体系结构方面的问题。
其次，需要从消息中提取数据。除了可以是文本、二进制、键值或其他形式的数据有效载荷外，还有其他系统和标头属性可以包含有用的信息。
不同的消息传递系统需要不同的API。除了具有自己的API的Kafka之外，大多数消息传递系统还支持JMS API或AMQP协议。
3.3.2 从Java Message Service系统收集数据
连接到Java消息服务（JMS）系统时，首先需要创建一个初始上下文，该上下文包含有关连接到提供程序的信息，例如代理URL和安全凭证。此信息因提供程序而异。 由此，您可以直接获得连接工厂，也可以通过Java命名和目录接口（JNDI）查找服务。 然后，工厂允许您创建与提供者的连接，并创建一个会话，通过该会话您可以发送和接收消息。
对于数据收集，感兴趣的是接收消息，这些消息可以来自队列，也可以来自主题。队列通常是点对点的，只有一个使用者会收到发送到队列的消息。主题提供了一种发布/订阅拓扑，每个订户都将收到一份已发布消息的副本。队列和主题在可伸缩性和可靠性方面各有各自的问题。
因为队列仅允许单个使用者接收消息的副本，所以不可能在不中断任何现有数据流的情况下将现有队列用作数据源。相反，需要添加其他队列（或主题）以及也路由到这些新目的地的现有消息。
从队列中读取具有传递保证，可以确保看到所有消息，但是这可能需要持久的选项来处理故障情况。主题更适合数据收集，因为它们可以有多个订阅者。但是，重要的是这些用户必须持久。这意味着消息将一直保留到每个订户都收到为止。否则它们将被丢弃。
收集JMS数据的最大问题是恢复。尽管JMS支持事务，但是它不允许在队列或主题内重新定位或倒退。在利用窗口或事件缓冲区的复杂的有状态处理管道中，恢复通常需要重播旧事件，而使用JMS API则不可能。
3.3.3 从Apache Kafka收集数据
Apache Kafka是一个高吞吐量的分布式消息传递系统。它利用了发布/订阅机制，并具有固有的持久性，将所有消息写入一个分布式提交日志。客户端以生产者或消费者的身份连接到代理，生产者向主题发送消息，消费者作为该主题的订阅者接收消息。当生产者发送消息时，它被存储在磁盘上的仅追加日志中。可以将代理聚集在大量的机器上，并在集群上对数据进行分区和复制。
当生产者向代理发送消息时，分区键用于确定需要将数据写入日志的分区，从而确定集群中的哪些机器需要将数据写入日志，每个分区写入一个单独的物理文件。出于可靠性和故障转移的目的，代理可以将数据写入一台或多台机器。日志将保留一段时间，使用者管理自己在日志中的读取位置。这使得消费者可以来去自如，以自己的速度运行，而不会影响到其他消费者。
使用者属于一个使用者组，组中的每个使用者被分配到一个或多个分区。订阅某个主题的每个使用者组将接收发送到该主题的所有消息，但是该组中的各个使用者将仅接收属于其分区的那些消息。不可能有比分区更多的使用者，因此决定主题的分区方案是一个基本的早期考虑。重要的是，因为每个使用者都需要跟踪它所读取的日志位置，所以使用者可以向后定位并重播旧的消息，只要它们保留在磁盘上。
在从Kafka收集数据时，同时考虑可伸缩性和可靠性是很重要的。
从Kafka读取数据的可伸缩性与为主题指定的分区数量直接相关。要使用多个使用者并行地从主题中读取数据，至少需要有与使用者相同数量的分区。以后可以向主题添加额外的分区，但这只影响新数据，而且不可能减少分区的数量。动态地将新的使用者添加到一个组(作为额外的线程或在独立的进程或机器中)，直到分区限制，这样就可以并行读取更多的数据。
Kafka与其他消息传递系统的主要区别在于，Kafka要求用户跟踪他们的读取位置。这有助于可靠性方面的考虑，因为在发生故障的情况下，使用者不仅可以从中断的地方恢复，而且还可以回退和重播旧的消息。通过跟踪使用者的读位置，并了解这些消息在处理管道中进行了多远的处理，可以确定使用者需要后退多远才能重新构建状态，然后才能继续处理。
3.3.4 处理不同的数据格式
前面描述的消息传递系统使用不同的方法来理解传输的数据。JMS支持多种类型的消息，包括原始字节、序列化的Java对象、文本和名称/值对。AMQP和Kafka本质上都是将数据作为原始字节发送，但是AMQP也可以以与HTTP一致的方式指定内容类型，而Kafka可以利用一个单独的模式注册表来定义主题上消息的数据结构。
然而，在大多数实际情况下，数据是文本序列化为字节，格式化为带分隔符的数据、日志文件条目、JSON或XML。从集合的角度来看，作为使用消息传递系统的一部分，启用文本(类似于文件)的灵活解析是很重要的。
3.4 云和API
越来越多的企业应用程序以SaaS多租户模式部署在云中。许多企业正在逐渐采用一种混合云部署模型，其中新的应用程序正在迁移到云中。
公司的所有业务应用程序很少会在单个公共云上运行。通常情况下，在整个运营和分析环境中，都会有一个跨越多个云和本地系统的计算网格。为了获得实时可见性，还需要以流方式提供来自这些云SaaS应用程序的数据。实际上，如果将本地系统设置为从本地应用程序接收流更改，则SaaS清单必须包括从SaaS环境实时获取数据的要求。
由于安全考虑（例如，某些网络端口的打开），服务级别协议（SLA）要求(CDC初始加载)，由于无法访问基础平台/数据库，我们在上一节中讨论的某些技术可能与SaaS环境不相关。或多租户可管理性问题（CDC的特殊触发器处理）通常，通过批量API批量提供业务对象的数据，或者通过流API实时提供业务对象的数据。
前面我们已经讨论过的一些技术的可能不属于SaaS环境因为无法理解的基础平台/数据库出于安全考虑(例如,某些网络端口的开放),服务水平协议(SLA)需求(CDC初始加载),或多租户可管理性（CDC的特殊触发器处理）的担忧。通常，业务对象的数据可以通过批量API批量提供，也可以通过流API实时提供。
3.5 设备和物联网
作为许多行业中数字化转型的重要推动力，物联网已经引起了广泛的关注。 简而言之，物联网是设备，传感器和执行器的全球集合，可以通过网络收集，传输和接收数据，而无需人工干预。 物联网中的“事物”可以指设备本身或它们正在监视的对象，包括人、动物、车辆、设备和机械。
尽管名称中提到了“互联网”，但物联网无需通过Web传输数据。此处的Internet是对Internet协议（IP）的引用，该协议允许仅基于IP地址将数据包从源传递到目的地。物联网可在任何基于IP的网络上运行，从而促进内部和外部使用案例。鉴于许多商业、工业和医疗用例将数据保密，因此物联网需要设备将数据传送到云是有限制性的。
3.5.1 从物联网设备收集数据
“IoT设备”涵盖了广泛的硬件。通过WiFi发送数据的单个温度传感器可以视为IoT设备。但是，包括温度传感器以及其他测量和逻辑的设备（例如智能恒温器，气象站或火灾警报器）也可以是IoT设备。 设备可以进一步组合以生产更大的“设备”，例如联网汽车、智能冰箱或家庭安全和控制系统。
设备可用的大小和电能将在某种程度上决定设备具有多少计算能力以及它可以支持的协议类型。较小的设备往往具有很少的内存或CPU功能，并且需要轻量级协议来传输数据。较大的设备可以执行更多处理，使用更复杂的代码，并支持重量更重，更具弹性的协议。
物联网使用的最简单的协议是TCP/IP网络模型的传输层上的TCP和UDP，将数据作为网络数据包直接发送到目的地。在应用层，可以使用现有协议，并且出现了新协议。HTTP和HTTPS（安全HTTP）是常见的，通常实现为通过代表性状态传输（REST）调用发送的JSON。
消息队列传输（MQTT）和WebSocket是常见的发布/订阅协议，允许与设备进行双向通信。OPC-UA（OPC基金会的OPC统一体系结构）是下一代标准，它定义了主要用于工业应用的客户端/服务器协议，利用UDP或MQTT在后台进行数据传输。
除了传输协议之外，另一个考虑因素是数据格式。物联网设备没有真正的标准，因此需要逐案考虑集成。JSON很常见，但是数据也可以是二进制、定界符、XML或以专有文本形式显示。
3.5.2 物联网可扩展性注意事项
关于物联网数据的任何讨论几乎总是包含边缘处理的概念。边缘处理是指计算尽可能靠近物理边缘设备（通常位于其中）使IoT设备尽可能“智能”。物联网在数据世界中产生的数据量相当独特，因为可能有成百上千甚至数百万个单独的设备都在生成少量数据。即使单个传感器或设备每秒仅生成10次数据，如果将其乘以设备数量，它也会很快变得不堪重负，其中许多数据是重复的，冗余的，或者只是没有那么有趣。该数据中真正需要的信息内容。
一个简单的例子是温度传感器。如果单个传感器每秒读取10次温度，它将每小时产生36,000个数据点。如果在此期间温度始终保持在70度，则信息内容为一项：“70度一小时”。
为了减少由IoT生成的数据量，可以通过单个边缘设备收集来自多个单独传感器的数据。在这里，可以对数据进行过滤，汇总和转换以提取信息内容。这里重要的是不仅要进行统计分析并发送摘要信息，而且还要能够对变化立即做出反应。大多数边缘处理是统计摘要加上即时更改检测和传感器运行状况信号的组合。
以温度传感器为例，它结合了特定时间段内的统计摘要（最小值，最大值，平均值，标准偏差等），并每分钟ping一次，以指示传感器是否还存在，并在温度变化剧烈时立即发出消息（超出平均值的两个标准差的正负）将大大减少需要收集的数据量。
可以在边缘进行的处理类型可能非常复杂，甚至可以结合机器学习和高级分析功能来快速响应重要事件。后面的章节中我们将对此进行详细讨论。
