从实时实时过滤和处理大量数据，到将日志数据和度量数据记录到不同来源的集中处理程序中，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