Kafka作为分布式消息队列,广泛应用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域,您可以使用Kafka数据源配置同步任务同步数据。本文为您介绍如何创建Kafka数据源。
背景信息
标准模式的工作空间支持数据源隔离功能,您可以分别添加并隔离开发环境和生产环境的数据源,以保护您的数据安全。详情请参见数据源开发和生产环境隔离。注意事项
支持阿里云Kafka,以及>=0.10.2且<=2.2.x的自建Kafka版本。
说明 对于<0.10.2版本Kafka,由于Kafka不支持检索分区数据offset,并且Kafka数据结构可能不支持时间戳,进而无法支持数据同步。
新增数据源
- 进入数据源管理页面。
- 登录DataWorks控制台。
- 在左侧导航栏,单击工作空间列表。
- 选择工作空间所在地域后,单击相应工作空间后的数据集成。
- 在左侧导航栏,单击 ,进入 页面。
- 在数据源管理页面,单击右上角的新增数据源。
- 在新增数据源对话框中,选择数据源类型为Kafka。
- 在新增Kafka数据源对话框中,配置各项参数。
- 可选:配置数据源的扩展参数。您可以选择为当前数据源配置扩展参数,即配置Kafka消费者和生产者的相关参数,格式为JSON格式。示例如下:
- 配置发往每个分区(Partition)的消息缓存量(消息内容的字节数总和)为16342。
- 配置每条消息在缓存中的最长时间为10毫秒。
{ "batch.size":"16342", "linger.ms":"10" }
配置项可选的压缩类型有:gzip、snappy、lz4。{ "compression.type":"gzip" }
说明 如果使用脚本模式配置的离线同步任务,或使用单表同步配置的实时同步任务中,配置的消费者或生产者参数,与扩展参数中配置的参数相同但取值不同,则扩展参数配置的参数优先级低于同步任务中配置的参数。 - 测试数据源与资源组的连通性。
- 测试连通性通过后,单击完成。