Kafka数据如何同步至MaxCompute之实践讲解

摘要：本次分享主要介绍Kafka产品的原理和使用方式，以及同步数据到MaxCompute的参数介绍、独享集成资源组与自定义资源组的使用背景和配置方式、Kafka同步数据到MaxCompute的开发到生产的总体部署操做等内容。

演讲嘉宾简介：耿江涛，阿里云智能技术支持工程师

如下内容根据演讲视频以及PPT整理而成。
本次分享主要围绕如下两个方面：

1、背景介绍
2、具体操做流程
1.Kafka消息队列使用以及原理
2.资源组介绍以及配置
3.同步过程及其注意事项 
4.开发测试以及生产部署

1、背景介绍
1. 实验目的
在平常工做中，不少企业将APP或网站产生的行为日志和业务数据经过Kafka收集以后作两方面的处理。一方面是离线处理，一方面是实时处理。而且通常会投递到MaxCompute中做为模型的构建，进行相关的业务处理，如用户的特征、销售排名、订单地区分布等。这些数据造成以后会在数据报表中做为展现。

2. 方案说明
Kafka数据同步到DataWorks有两条链路。一条链路是业务数据和行为日志经过Kafka，再经过Flume 上传到Datahub，以及Max Compute，最终在QuickBI进行展现。另外一条链路是业务数据和行为日志经过Kafka以及DataWorks，MaxCompute，最终在QuickBI当中展现。
本次展现Kafka经过DataWorks上传到MaxCompute的流程。从DataWorks上传到MaxCompute是经过两种方案进行上传数据同步的。方案一是自定义资源组，方案二是独享资源组。自定义资源组通常适用于复杂网络的数据上云场景。独享资源组操做方式主要针对集成资源不足的状况。

2、具体操做流程
1．Kafka消息队列使用及其原理
Kafka产品概述：消息队列 for Apache Kafka 是阿里云提供的分布式、高吞吐、可扩展的消息队列服务。消息队列for Apache Kafka通常用于日志收集、监控数据聚合、流式数据处理、在线离线分析等大数据领域。消息队列 for Apache Kafka 针对开源的 Apache Kafka 提供全托管服务，完全解决开源产品长期以来的痛点。云上Kafka具备低成本、更弹性、更可靠的优点，用户只需专一于业务开发，无需部署运维。

Kafka架构介绍：以下图所示，一个典型的Kafka集群主要分为四部分。Producer生产数据并经过 push 模式向消息队列 for Apache Kafka 的 Kafka Broker 发送消息。发送的消息能够是网站的页面访问、服务器日志，也能够是 CPU 和内存相关的系统资源信息。Kafka Broker用于存储消息的服务器。Kafka Broker 支持水平扩展。 Kafka Broker 节点的数量越多，Kafka 集群的吞吐率越高。Kafka Broker针对topic会partition一个概念，partition有leader、follower的角色分配。Consumer经过 pull 模式从消息队列 for Apache Kafka Broker 订阅并消费leader的信息数据。其中partition内部有offset做为消息的消费点位。经过ZooKeeper管理集群的配置、选举 leader 分区，而且在Consumer Group 发生变化时，管理partition_leader的负载均衡。

Kafka消息队列购买以及部署：以下图，用户首先能够到Kafka消息队列产品页面点击购买，根据我的状况选择对应包年、包月等消费方式、地区、实例类型、磁盘、流量以及消息存放时间。其中较为重要的一点是要选择对应地区，若是用户的MaxCompute在华北，那么尽可能选择华北地区。选择开通完成后须要进行部署。点击部署，选择合适的VPC及其交换机进行部署。

部署完成后进入Kafka Topic管理页面，点击建立Topic输入本身的Topic。Topic命名下面有三条注意信息，命名尽可能跟本身的业务一致，好比是财经业务或者是商务业务，尽可能进行区分。第四步进入Consumer Group管理，点击建立Consumer Group建立本身所须要的Consumer Group。Consumer Group的命名也须要规范，若是是财经或商务业务，尽可能和本身的Topic相对应。

Kafka白名单配置：Kafka安装部署完成以后确认须要访问Kafka的服务器或产品的白名单。下图中的默认接入点即为访问接口。

2．资源组介绍及其配置
自定义资源组的使用背景：自定义资源组通常针对IDC之间的网络问题。本地网络和云上网络存在差别，如DataWorks能够经过免费传输能力（默认任务资源组）进行海量数据上云，但默认资源组没法实现传输速度存在较高要求或复杂环境中的数据源同步上云的需求。此时用户可使用自定义资源组可实现复杂环境同步上云的需求，解决DataWorks默 认资源组与您的数据源不通的问题，或实现更高速度的传输能力。然而，自定义资源组主要解决的仍是复杂网络环境上云同步问题，打通任意网络环境之间的数据传输同步。

自定义资源组的配置：自定义资源组的配置须要六步操做，首先点击进入DataWorks控制台，点开工做空间的列表，选择用户须要的项目空间，点击进入数据集成，即确认本身的数据集成是要在哪一个空间项目下进行添加。以后，点击进入数据源界面，点击新增自定义资源组。要注意页面右上角的新增自定义资源组是只有项目管理员有权限添加。

第三步是确认Kafka与须要添加的自定义资源组属于同一个VPC下。本次实验是ECS向Kafka发送消息，两者的VPC应该一致。第四步登陆ECS，即我的的自定义资源组。执行命令dmidecode|grep UUID获得ECS的UUID。

第五步是将添加服务器UUID以及自定义资源组的IP或机器CPU和内存填写进来。最后是在ECS上执行相关命令，Agent安装共5步，作一一确认，在第4小步完成后点击刷新查看服务是否为可用状态。添加完成后进行检查连通测试，检查是否添加成功。

独享资源组的使用背景：一些客户反映在Kafka同步到MaxCompute时会报资源不足的问题，能够经过新增独享资源组的方式进行数据同步。独享资源模式下，机器的物理资源（网络、磁盘、CPU和内存等）彻底独享。不只能够隔离用户间的资源使用，也能够隔离不一样工做空间任务的资源使用。此外，独享资源也支持灵活的扩容、缩容功能，能够知足资源独 
享、灵活配置等需求。独享资源组能够访问在同一地域下的VPC数据源，同时也能够访问跨地域的公网RDS地址。

独享资源组的配置：独享资源组的配置主要须要两步操做，首先进入DataWorks控制台的资源列表，点击新增独享资源组，包括独享集成资源组和独享调度资源组。此处选择新增独享集成资源组，点击购买时仍要注意选择对应的购买方式、区域、资源、内存、时间期限、数量等。

购买完成后须要把独享集成资源组绑定到与Kafka对应的VPC，点击专有网络绑定，选择与Kafka对应的交换机（最明显的是可用区的区别）、安全组。

3．同步过程及其注意事项
Kafka同步到MaxCompute的须要进行相关参数配置同时须要注意如下几个事项。
DataWorks数据集成操做：进入DataWorks操做界面，点击建立业务流程，在新建的业务流程添加数据同步节点，再进行命名。

以下图所示，进入数据同步节点，包括Reader端和Writer端，点击Reader端数据源为Kafka，Writer端数据源为ODPS。点击转化为脚本模式。下图右上角是帮助文档，Reader或Writer端的一些同步参数能够在此处就近点击，方便阅读、操做和理解。

Kafka Reader的主要参数：Kafka Reader的主要参数首先server，上文所述Kafka的默认接入点就是其中一个server，ip:port。注意此处server是必填参数。topic，表示在Kafka部署完成以后，Kafka处理数据源的topic，此处也是必填参数。下一个参数是针对列column，column支持常量列、数据列、属性列。常量列和数据列不过重要。同步的完整消息通常存放在属性列 value 中，若是须要其它信息，如partition、offset、timestamp，也能够在属性列中筛选。column是必填参数。

keyType、valueType各有6种类型，根据用户同步的数据，选择相应的信息，同步一个类型。须要注意同步方式是按消息时间同步，仍是按消费点位置同步的。按数据消费点位置同步有四个场景，beginDateTime，endDateTime，beginOffset，endOffset。 beginDateTime 和beginOffset 二选其一，做为数据消费起点。endDateTime 和endOffset 二选其一。须要注意beginDateTime、endDateTime 中须要Kafka0.10.2版本以上才支持按数据消费点位置同步功能。另外须要注意beginOffset有三个比较特殊的形式：seekToBeginning，表示从开始点位消费数据；seekToLast，表示从上次消费的偏移位置消费数据，按照beginOffset从上次偏移位置只能一次消费，若是使用beginDateTime则能够屡次消费，这取决于消息存放时间；seekToEnd，表示从最后点位消费数据，会读取到空数据。

skipExceeedRecord没有太大做用，是没必要填项。partition对topic全部分区共同读消费的，因此无需自定义一个分区，是非必填项。kafkaConfig，若是有其它相关配置参数能够扩展配置在kafkaConfig，kafkaConfig也是非必填项。

MaxCompute Writer的主要参数：dataSource是数据源名称，添加ODPS数据源。tables，表示所建立的数据表的表名称，Kafka的数据要同步到哪张表中，相应的字段也能够创建。
partition，若是表为分区表，则必须配置到最后一级分区，肯定同步位置。若为非分区表，则没必要填。column，尽可能与Kafka column中的相关字段作一一对应的操做。同步的字段对应，信息同步才能确认成功。truncate，写入时同步的数据是选择以追加模式写仍是以覆盖模式写，尽可能避免多个DDL同时操做一个分区，或者在多个并发做业启动前提早建立分区。

Kafka同步数据到MaxCompute：将下图拆分为三部分。Kafka的Reader端，MaxCompute的Writer端以及限制参数。Reader包含server、endOffset、kafkaConfig、group.id、valueType、ByteArray、column字段、topic、beginOffset、seekToLast等。MaxCompute的Writer端包含覆盖、追加、压缩、查看源码、同步到的表、字段要和Kafka的Reader端作一一对应，最重要的是value数据同步。限制参数，主要有errorlimit，数据超过几个错误后会进行报错；speed，能够限制流速、并发度等。

参考Kafka生产者SDK编写代码：最终生产出的数据要发送到Kafka中，经过相关代码能够查看用户的生产数据。下图一段代码表示配置信息的读取，协议、序列化方式以及请求的等待时间，须要发送哪个topic，发送什么样的消息。发送完成后回传一个信息。详细代码能够参考配置文件、消息来源、生产者消费者的代码模板: 
https://help.aliyun.com/document_detail/99957.html?spm=a2c4g.11186623.6.566.45fc54eayX69b0。

代码打包运行在ECS上（与Kafka同一个可用区）：以下图所示，执行crontab-e命令，每到17:00执行一次。下图为发送日志完成后的消息记录。

在MaxCompute上建立表：进入DataWorks业务流程页面，建立目标表，使用一个DDL语句建立同步的表，或根据用户我的业务相应建立不一样的表的字段。

4．开发测试以及生产部署
选择自定义资源组（或独享集成资源组）进行同步操做：下图所示，选择右上角“配置任务资源组”，根据用户我的需求选择资源组，点击执行。执行完成后，会出现标识显示成功，同步数据记录以及结果是否成功。同步过程基本结束。

查询同步的数据结果：在DataWorks临界面查看同步结果，在临时节点点击查询命令，select * from testkafka3（表），查看数据同步结果。数据已经同步过来，证实测试成功。

设置调度参数：业务流程开发数据同步以后，会对相关模型进行一些业务处理，最后设计一些SQL节点、同步节点，进行部署。以下图所示，在右侧点击调度配置，输入调度时间。具体操做可参考DataWorks官方文档完善业务处理流程。

提交业务流程节点，并打包发布：点击业务流程，选择所须要提交的节点并提交。一些业务流程提交以后不须要放到生产环境当中。而后进入任务发布界面，将节点添加到待发布进行任务部署。

确认业务流程发布成功：最后在运维中心页面，确认发布是否在生产环境中存在。至此Kafka同步数据到MaxCompute过程结束。到了对应的调度时间，在各个节点或者右上角会有节点的日志展现，能够查看日志运行状况是否正常，或是否须要进行后续操做，部署数据或是相关命令。

 

 

原文连接

本文为云栖社区原创内容，未经容许不得转载。