通过Apache Kafka集成流式传输大数据

从实时实时过滤和处理大量数据，到将日志数据和度量数据记录到不同来源的集中处理程序中，Apache Kafka日益集成到各种系统和解决方案中。使用CData Sync ，可以轻松地将此类解决方案应用于任何CRM，ERP或Analytics软件。

配置Apache Kafka目标以进行CData同步

在CData Sync中设置Kafka目标非常简单。只需提供“服务器”和“端口”，然后复制命令就可以接管-不需要其他配置。要设置这种连接，请首先导航到“连接”页面，然后单击“目标”选项卡，然后选择“ Kafka”。

指定服务器和端口属性。如果启用了身份验证，请同时指定“用户”和“密码”属性。单击“保存更改”和“测试连接”以保存更改，并确保CData Sync可以连接到Kafka服务器。

其他一些属性可用，并在“高级”选项卡中进行了分类：

启用幂等：确保邮件仅传递一次。在某些情况下，生产者可能会产生重复的消息。为了进行验证，客户端可以在执行作业后将消耗的结果数与“受影响的记录”状态进行比较。
序列化格式：指定产生的消息的格式；可用值为JSON，XML和CSV 。
主题：如果指定，该属性将覆盖表名称作为复制的目标主题。

增量更新如何工作

CData Sync使无缝更新变得无缝。服务器不需要配置，并且所有必需的属性都已预先配置，具体取决于Source和Source表。

与SQL Server等其他数据库工具不同，Kafka不支持可靠的状态存储方式。 CData Sync使用本地SQLite数据库来解决此问题。它将存储上次复制表的时间，并使用该时间戳过滤最新记录。大多数企业系统都提供一个系统列来指定记录的最后更新日期，这足以满足此目的。

例如，QuickBooks Online中的“帐户”表包含此类列。复制表：

并在修改三个记录后运行另一个副本：

某些表没有自动更新的列，该列保存记录的最后更新日期。在这种情况下，别无选择，只能从一开始就完全复制结果。 Kafka提供了附加到消息的时间戳字段，可用于区分较新的结果。

QuickBooks Online中的department表没有用于指定上次更新时间的列。复制此表将导致：

在添加两个新记录的同时再运行一次复制会产生：

优化查询

有多种方法可以管理CData Sync生成的消息大小。根据Kafka服务器的配置，可能需要进行优化，或者仅考虑使复制性能有所提高就值得考虑。

压缩类型：指定如何压缩生成的数据。可用选项为gzip，lz4，snappy或无。指定非“ none”以外的压缩类型将减少消息有效负载。

最大批处理大小：指定在单个请求中发送的最大批处理大小（以字节为单位）。批处理中充满了整个消息。如果批次已等待一段时间，则可以提前发送批次而无需填充。降低此值可能会降低性能，但是如果生成的消息超过服务器允许的最大消息大小，则可能有必要。

排除列：如果单个记录本身太大，则转换功能提供了一种从输出消息中省略某些列的方法。这是最常用的聚合列。要排除列，请导航至作业，然后单击所需表旁边的转换按钮：

接下来，取消选择聚合列：

最后，单击“确定”进行保存。

设置CData Sync来管理数据源

通过使用计划作业，可以设置完全自动的记录提取，Kafka消费者可以使用它来始终获取新条目的最新信息。时间可以根据特定数据集的需要进行调整。

要安排作业，请选择所需的作业。在“计划”部分下，选中“计划此作业自动运行”框。最后，选择适当的间隔。

结论

CData Sync和Apache Kafka是强大的组合；可以从任何数据源复制到Kafka使用者，以支持从分析到日志记录的一系列需求。自动检测新记录并安排作业可确保新数据稳定地流向其订户。压缩，转换和其他优化可进一步控制数据格式，量和频率。下载CData Sync的30天免费试用版，今天就开始将大数据流式传输到Apache Kafka！

翻译自: https://www.javacodegeeks.com/2019/11/stream-big-data-with-apache-kafka-integration.html