利用变更数据采集（CDC）同步数据库（2）

　　这是关于变更数据采集（CDC）系列的第二部分。在这篇文章中，让我们讨论CDC的使用案例，并看看有哪些工具可以帮助你把CDC添加到你的架构中。
　　变更数据采集有助于实现事件驱动的应用架构。它允许应用程序监听数据库、数据仓库等的变化，并根据这些变化采取行动。
　　在高层次上，以下是对数据变化采取行动所产生的用例和架构。 提取、转换、加载（ETL）。 捕获一个数据存储的每一个变化，并将这些变化应用于另一个数据存储，可以实现复制（一次性同步）和镜像（连续同步）。  集成和自动化。 对数据变化事件采取的行动可以使任务自动化，触发工作流程，甚至执行云功能。  历史。 当对一个数据集进行历史分析时，拥有数据的当前状态和所有过去的变化为你提供完整的信息，以进行更高的保真分析。  提醒。 大多数情况下，只要用户关心的数据发生变化，应用程序就会向用户发送一个事件。CDC可以成为实时警报系统的触发器。
　　我们来探讨一下。 提取、转换、加载
　　到目前为止，CDC最常见的用例之一是提取、转换、加载（ETL）。ETL是一个过程，在这个过程中，你从一个源头获取数据（提取），以某种方式处理它（转换），并将其发送到一个目的地（加载）。
　　数据复制（一次性同步）和镜像（连续复制）是ETL过程的很好例子。ETL是一个总括性的术语，包含了非常不同的使用情况，如：。 将数据从数据库输入到数据仓库，以运行分析查询而不影响生产。  保持缓存和搜索索引系统的最新状态
　　CDC不仅可以帮助解决这些用例，而且也是解决这些问题的最佳方式。例如，为了将数据镜像到数据仓库，你必须在源数据库发生任何 变化 时捕捉并应用这些 变化 。正如本系列第一部分所讨论的 Streaming Replication Logs系列的第一部分所讨论的，CDC被数据库用来保持备用实例的最新状态以进行故障转移，因为它是有效和可扩展的。当在一个更广泛的架构中挖掘这些事件时，你的数据仓库可以像用于灾难恢复的备用数据库实例一样保持最新。
　　保持 caches和搜索索引系统的更新，也是ETL问题和CDC的伟大用例。今天创建的大型应用程序是由许多不同的数据存储组成的。例如，某些架构会利用Postgres、Redis和Elasticsearch作为关系型数据库、缓存层和搜索引擎。所有这些都是为特定的数据使用案例而设计的记录系统，但数据需要在每个存储中进行镜像。
　　你永远不希望用户在搜索一个产品时发现它不再存在。陈旧的缓存和搜索索引会导致可怕的用户体验。CDC可以用来建立数据管道，使这些存储与它们的上游依赖关系保持同步。
　　理论上，一个应用程序可以同时向Postgres、Redis和Elasticsearch写数据，但 ＂双写 ＂的管理很困难，可能导致系统不同步。CDC提供了一个更强大、更容易维护的实现。与其将更新索引和缓存的逻辑添加到一个单一的单体应用中，不如创建一个事件驱动的微服务，可以独立于面向用户的系统进行构建、维护、改进和部署。这个微服务可以保持索引和缓存的更新，以确保用户在最相关的数据上操作。 集成和自动化
　　SaaS的兴起使产生数据或需要更新数据的工具数量激增。CDC可以提供一个更好的模型来保持Salesforce、Hubspot等的更新，并允许对需要响应这些数据变化的业务逻辑进行自动化。
　　我们上面描述的每个用例都将数据发送到一个特定的目的地。然而，最强大的目的地是一个云功能。捕获数据变化并触发云函数可以用来执行本文提到的（或没有提到的）每个用例。
　　云功能已经得到了巨大的发展，因为没有服务器需要维护；它们会自动扩展，而且使用和部署都很简单。这种流行性和实用性已经在JAMStack等架构中得到了明显的证明。CDC与这种架构模式完美契合。
　　今天，云功能是由一个事件触发的。这个事件可能是当一个 file is uploaded to Amazon S3或一个HTTP请求。然而，正如你可能已经猜到的，这个触发事件可能是由CDC系统发出的。
　　例如，这里有一个AWS Lambda函数来接受一个数据变化事件，并且进行Algolia的搜索数据索引: const algoliasearch = require(＂algoliasearch＂); const client = algoliasearch(process.env.ALGOLIA_APP_ID, process.env.ALGOLIA_API_KEY); const index = client.initIndex(process.env.ALGOLIA_INDEX_NAME); exports.handler =  async function(event, context) {     console.log(＂EVENT: n＂ + JSON.stringify(event, null, 2))     const request = event.Records[0].cf.request;     // Accessing the Data Record     //      const body = Buffer.from(request.body.data, ＂base64＂).toString();     const { schema, payload } = body;     const { before, after, source, op } = payload;     if (req.method === ＂POST＂) {         try {             // if read, create, or update operation create o update index             if (op === ＂r＂ || op === ＂c＂ || op === ＂u＂) {                 console.log(`operation: ${op}, id: ${after.id}`)                 after.objectID = after.id                 await index.saveObject(after)             } else if (op === ＂d＂) {                 console.log(`operation: d, id: ${before.id}`)                 await index.deleteObject(before.id)             }             return res.status(200).send()         } catch (error) {             console.log(`error: ${JSON.stringify(error)}`)             return res.status(500).send()         }     }     return context.logStreamName }
　　每次触发这个函数，它将查看数据变化（op），并在Algolia中执行相应的操作。例如，如果一个删除操作发生在数据库中，我们可以在Algolia中执行一个 deleteObject 在Algolia中。
　　响应 CDC 事件的函数可以是小而简单的。但是，CDC--与基于事件的架构一起--也可以简化原本非常复杂的架构。
　　例如，在你的应用程序中实现Webhooks的功能，在CDC中成为一个更直接的问题。Webhooks允许用户在某些事件发生时触发一个POST请求，通常是数据变化。例如，用 Github，你可以在一个拉动请求被合并时触发一个云函数。合并的拉动请求是对数据存储的UPDATE操作，这意味着CDC系统可以捕获这个事件。一般来说，大多数 webhook 事件可以转化为 CDC 系统可以捕获的 INSERT UPDATE 和 DELETE 操作。 历史
　　在某些情况下，你可能不想对CDC事件采取行动，而只想存储原始变化。使用CDC，数据管道可以将所有的变化事件存储到云桶中，以便长期处理和分析。为历史分析而存储数据的最佳地点是在一个云桶内，被称为数据湖。
　　数据湖是一个集中式存储，允许你以任何规模存储所有的结构化和非结构化数据。数据湖通常利用云对象桶解决方案，如亚马逊S3或 Digital Ocean Spaces.
　　例如，一旦数据在数据湖中，SQL查询引擎如 Amazon Presto可以针对变化的数据集运行分析查询。
　　在存储原始变化的同时，你不仅有数据的当前状态，你还有 所有 以前的状态（历史）。这就是为什么CDC为历史分析增加了大量的价值。
　　拥有历史数据可以让你支持灾难恢复工作，也可以让你回答关于数据的追溯问题。例如，假设你的团队重新定义了月活跃用户（MAU）的计算方式。有了用户数据集的完整历史，就可以根据过去的任何日期进行新的MAU计算，并将结果与当前的状态进行比较。
　　这种丰富的历史也有面向用户的价值。审计日志和活动日志是向用户显示数据变化的功能。
　　捕获和存储变化事件在实现这些功能时提供了一个更好的架构。与Webhooks一样，审计日志和活动日志的根源在于CDC系统可以捕获的操作。 告警
　　任何警报系统的工作都是为了通知利益相关者一个事件。例如，当你收到一个新的电子邮件通知时，你会被通知到一个电子邮件数据存储的INSERT操作。通常情况下，大多数警报都与数据存储中的变化有关，这意味着CDC很适合为警报系统提供动力。
　　例如，假设你有一个电子商务商店。在对一个采购表启用CDC后，你可以捕获变化事件，并在有新的采购时通过执行Slack警报来通知团队。
　　就像审计或活动日志一样，由CDC提供的通知不仅可以提供有关发生的事件的信息，还可以提供变化本身的细节。
　　汤姆将标题从 ＂会议记录 ＂更新为 ＂我的新会议＂。
　　这种警报行为也有内部价值。从基础设施监控的角度来看，CDC 事件可以提供对用户如何与您的应用程序和数据互动的洞察力。例如，你可以看到用户何时以及如何添加、更新或删除信息。这些数据可以被发送到 Prometheus UI来监控这些信息并采取行动。 开始使用CDC
　　在 第一部分中，我们谈到了CDC的各种常见实施方式。  轮询  数据库触发器  流日志处理
　　这些都可以用来构建我们在本文中讨论的用例。最重要的是，由于CDC关注的是数据，这个过程与编程语言无关，可以集成到大多数架构中。
　　轮询和触发器
　　当使用轮询或数据库触发器时，没有开销，也不需要安装。你可以通过建立你的查询来轮询或利用你的数据库的触发器（如果它们被支持）来开始。
　　流日志处理
　　数据库使用流式复制日志进行备份和恢复，这意味着大多数数据库开箱即提供一些CDC行为。挖掘这些事件的难易程度取决于数据存储本身。开始的最好地方是挖掘你的数据库的复制功能。下面是一些最流行的数据库的复制日志资源。  PostgreSQL’s Write-Ahead Logs  MySQL’s Binary Log  MongoDB’s Oplog  CockroachDB
　　要开始使用流式日志，答案是与有关的数据库紧密相连的。在今后的文章中，我将探讨每一种情况下是什么样的。
　　直接实施其中的任何一项确实需要一些时间、计划和努力。如果你想开始使用CDC，最低的门槛是采用一个CDC工具，它知道如何从你使用的数据存储中沟通和捕捉变化。 更改数据采集工具
　　这里有一些很好的工具供你评估。
　　Debezium
　　Debezium是迄今为止最流行的CDC工具。它维护得很好，是开源的，建立在 Apache Kafka.它支持 MongoDB,MySQLPostgreSQL和更多的数据库，开箱即用。
　　在高层次上，Debezium使用Hook数据库的复制日志，并将变化事件发送到Kafka。你甚至可以运行 Debezium standalone不使用Kafka。
　　真正好的是，Debezium都是基于配置的。在安装和配置Debezium后，你可以使用基于JSON的配置来配置与数据存储的连接。 {   ＂name＂: ＂fulfillment-connector＂,     ＂config＂: {     ＂connector.class＂: ＂io.debezium.connector.postgresql.PostgresConnector＂,      ＂database.hostname＂: ＂192.168.99.100＂,      ＂database.port＂: ＂5432＂,      ＂database.user＂: ＂postgres＂,      ＂database.password＂: ＂postgres＂,      ＂database.dbname＂ : ＂postgres＂,      ＂database.server.name＂: ＂fulfillment＂,      ＂table.include.list＂: ＂public.inventory＂    } }
　　一旦连接，Debezium会对你的数据进行初始快照，并向Kafka主题发出变化事件。然后，服务可以 consume the topics并对其采取行动。
　　这里有一些可以开始使用Debeizium的好地方。  Debezium resources on the web  Debezium Tutorial
　　Meroxa
　　Meroxa是一个实时数据协调平台，为您提供实时基础设施。Meroxa消除了与配置和管理经纪人、连接器、转换、功能和流媒体基础设施相关的时间和开销。您所要做的就是添加您的资源和构建您的管道。Meroxa支持 PostgreSQL, MongoDB, SQL Server，以及更多。
　　CDC管线可以在可视化仪表板中建立，也可以使用 Meroxa CLI: # Add Resource $ meroxa resource add my-postgres --type postgres -u postgres://$PG_USER:$PG_PASS@$PG_URL:$PG_PORT/$PG_DB # Add Webhook $ meroxa resource add my-url --type url -u $CUSTOM_HTTP_URL # Create CDC Pipeline $ meroxa connect --from my-postgres --input $TABLE_NAME --to my-url
　　上面的用例有更详细的介绍。
　　有一些很好的资源来帮助你开始使用Meroxa。
　　我迫不及待地想看到你建造的东西。