开发者学堂课程【SaaS 模式云数据仓库系列课程 —— 2021数仓必修课：Kafka 数据如何同步到 MaxCompute】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/55/detail/1039
Kafka 数据如何同步到 MaxCompute
目录：
l 实验目的
l 方案说明
l Kafka 消息队列使用以及原理
l 资源组介绍以及配置
l 同步过程及其注意事项
l 开发测试以及生产部署
实验目的 :
日常工作中,企业需要将 APP 或网站产生的行为日志和业务数据,通过 Kafka 消息队列统一收集后,投递到数据仓库 MaxCompute 中,再通过大数据分析后将指标数据在报表中展示,如用户特征、销售排名、订单地区分布等。
通过本次实验,我们可以学习了解 Kafka数据如何通过 Dataworks 数据集成同步到 MaxCompute ;
方案说明 :
方案一,使用自定义资源组的背景一般为网络环境复杂适用于数据上云的场景,该实验将展示使 CS 作为自定义资源组的操作过程。
方案二,使用独享集成资源组背景一般为集成资源不足影响数据同步过程,该实验将展示使用独享资源组的操作方式
(重点关注 VPC 的绑定).
一．Kafka 消息队列使用以及原理
1、Kafka 产 品概述
消息队列 for Apache Kafka 是阿里云提供的分布式、高吞吐、可扩展的消息队列服务。消息队列 for Apache Kafka 广泛用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域。
消息队列 for Apache Kafka 针对开源的Apache Kafka 提供全托管服务,彻底解决开源产品长期以来的痛点。有了消息队列 for Apache Kafka ,您只需专注于业务开发,无需部署运维,具有低成本、更弹性、更可靠的优势。
2、Kafka 架构介绍
一个典型的消息队列 for Apache Kafka 集群包括四个部分:
(1) Producer :通过 push 模式向消息队列 for Apache Kafka 的 Kafka Broker 发送消息。发送的消息可以是网站自官自 的页面访问、服务器日志,也可以是 CPU 和内存相关的系统资源信息。
(2) Kafka Broker :用于存储消息的服务器。 Kafka Broker 支持水平扩展。 Kafka Broker 节点的数量越多, Kafka 集群的吞吐率越高。
(3) Consumer Group :通过 pul 模式从消息队列 for Apache Kafka Broker 订阅并消费消息。
(4) Zookeeper :管理集群的配置、选举 leader 分区,并且在 Consumer Group 发生变化时,进行负载均衡。
3.Kafka 消息队列购买以及部署
1) 到 Kafka 消息队列产品页面点击购买,选择对应消费方式,地区,实例类型,磁盘,流量以及息存放时间。
2) 开通完成之后点击部署,选择合适的 VPC 以及交 换机注意可用区的位置。
3) 进入 Topic 管理页面,点击创建 Topic 按钮，创建个人的 Topic . Group .
4) 进入 Consumer Group 管理,点击创建 Consumer Group ，创建自己所需的 Consumer Group .
4.Kafka 白名单配置
确认需要访问需要访问 Kafka 的网段信息。
二．资源组介绍以及配置
1. 自定义资源组的使用背景
DataWorks 可以通过免费传输能力(默认任务资源组)进行海量数据上云,但默认资源组无法实现传输速度存在较高要求或复杂环境中的数据源同步上云的需求。您可以新增自定义的任务资源运行数据同步任务,解决DataWorks 默认资源组与您的数据源不通的问题,或实现更高速度的传输能力。
当默认任务资源无法与您的复杂的网络环境连通时,可以通过数据集成自定义资源的部署,打通任意网络环境之间的数据传输同步。
2.自定义资源组的配置
1) 进入 Dataworks 控制台,点击需要数据同步的项目空间,点击数据集成。
2) 进入数据源界面,点击新増自定义资租。
3) 确认 Kafka 与需要添加自定义资源组属于同一个 VPC下.
4) 登录 ECs ,执行命令 dmidecode I grep UUID 得到ECs 的 UUID .
5) 将 ECS 的 UUID 以及护,所占用资源的 CPU 与内存填写进来。
6) 在 ECS 上执行安装 Agent 的命令,添加完成测试连通性。
3.独享资源组的使用背景
独享资源模式下,机器的物理资源(网络、磁盘、 CPU 和内存等)完全独享。不仅可以隔离用户间的资源使用,也可以隔离不同工作空间任务的资源使用。此外,独享资源也支持灵活的扩容、缩容功能,可以满足资源独享、灵活配置等需求。
独享资源组可以访问同一地域的 VPC 数据源,也可以访问跨地域的公网 RDS 地址。
7) 进入 DataWorks 控制台的资源组列表,点击新增独享集成资源组,点击购买选择对应的地区, CPU 以及内存。
8) 点击专有网路绑定,选择与 Kafka 对应 VPC 以及交换机明显的区别是可用区),安全组。
三．同步过程及其注意事项
1. DataWorks 数据集成操作
1) 进入 DataWorks 操作界面,点击创建业务流程,在新建的业务流程里添加数据同步节点。
2) 进入数据同步节点,点击数据源为 Kafka ,点击转化为脚本模式。
2. Kafka Reader的主要参数讲解
3.Kafka同步数据到MaxCompute
4.参考Kafka生产者SDK编写代码
详细代码参考文档涉及到配置文件,消息来源,生产者消费者的代码模板:
https://help.alivm.com/document detail/9995 7 .html?spma2c4g.11186623.6.566.45fc54eaxX69b0
5.代码打包运行之在ECS上（与Kafka同一个可用区）
1.执行 crontab - e 执行定时任务发送消息
017***java - jar home / expot/ upload/javaCode З. jar >> home / export / upload /logfile . log
2.查看发送消息的定时任务日志
6.在 MaxCompute 上创建表
1.创建目标表界面
2.DDL语句
CREATE TABLE ` testkafka З`(`
value ` string ,
` timestampl ` string ,[ partition ` string , offset ` string
));
四．开发测试以及生产部署
1.选择自定义资源组(或独享集成资源组)进行同步操作
① 选择可使用的独享资源组与自定义资源组进行同步。
② 同步任务成功会显示,同步数据记录以及结果标志。
2.查询同步的数据结果
在 Datalorks 的临时界面查看同步数据结果。
3.设置调度参数
① 点击右侧调度配置,输入调入时间。
② 参考 Datalork 官方文档完善业务处理流程。
4.业务流程节点,并打包发布
① 点击业务流程,提交业务流程节点。
② 进入任务发布界面,将节点添加到特发布进行任务的部署。
5.确认业务流程发布成功
在运维中心页面,确认发布是否在生产环境中存在,至此 Kafka 同步数据到 MaxCompute 过程结束。