开发者学堂课程【SaaS 模式云数据仓库实战：Kafka 数据如何同步到 MaxCompute】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/332/detail/3718
Kafka 数据如何同步到 MaxCompute
内容介绍：
一、Kafka 消息队列使用以及原理
二、资源组介绍以及配置
三、同步过程及其注意事项
四、开发测试以及生产部署
实验目的：
日常工作中，企业需要将 APP 或网站产生的行为日志和业务数据，通过 Kafka 消息队列统一收集后，投递到数据仓库 MaxCompute 中，再通过大数据分析后将指标数据在报表中展示，如用户特征、销售排名、订单地区分布等。
通过本次实验，我们可以学习了解 Kafka 数据如何通过 Dataworks 数据集成同步到 MaxCompute。
接下来我们所要展示的是 Kafka 通过 Dataworks 上传到 MaxCompute，其中是通过两种方案进行上传的。
方案一，使用自定义资源组的背景一般为网络环境复杂适用于数据上云的场景，该实验将展示使用 ECS 作为自定义资源组的操作过程。
方案二，使用独享集成资源组背景一般为集成资源不足影响数据同步步过程，该实验将展示使用独享资源组的操作方式(重点关注 VPC 的绑定)。
一、Kafka 消息队列使用以及原理
1、Kafka 产品概述
消息队列 for Apache Kafka 是阿里云提供的分布式、高吞吐、可扩展的消息队列服务。
消息队列 for Apache Kafka 广泛用于日志收集、监控数据聚合、流式数据处理、在线和离线分析等大数据领域。
消息队列 for Apache Kafka 针对开源的 Apache Kafka 提供全托管服务，彻底解决开源产品长期以来的痛点。
有了消息队列 for Apache Kafka，您只需专注于业务开发，无需部署运维，具有低成本、更弹性、更可靠的优势。
2、Kafka 架构介绍
一个典型的消息队列 for Apache Kafka 集群包括四个部分：
Producer：通过 push 模式向消息队列 for Apache Kafka 的 Kafka Broker 发送消息。发送的消息可以是网站的页面访问、服务器日志，也可以是 CPU 和内存相关的系统资源信息。
Kafka Broker：用于存储消息的服务器。Kafka Broker 支持水平扩展。Kafka Broker 节点的数量越多，Kafka 集群的吞吐率越高。
Consumer Group：通过 pull 模式从消息队列 for Apache Kafka Broker 订阅并消费消息。
Zookeeper：管理集群的配置、选举 leader 分区，并且在 Consumer Group 发生变化时，进行负载均衡。
3、Kafka 消息队列购买以及部署
(1)到 Kafka 消息队列产品页面点击购买，选择对应消费方式，地区，实例类型，磁盘，流量以及消息存放时间。
(2)开通完成之后点击部署，选择合适的 VPC 以及交换机(注意可用区的位置)。
(3)进入 Topic 管理页面，点击创建 Topic 按钮，创建个人的 Topic。
(4)进入 Consumer Group 管理，点击创建 Consumer Group，创建自己所需的 Consumer Group。
4、Kafka 白名单配置
确认需要访问 Kafka 的网段信息。
二、资源组介绍以及配置
1、自定义资源组的使用背景
DataWorks 可以通过免费传输能力(默认任务资源组)进行海量数据上云，但默认资源组无法实现传输速度存在较高要求或复杂环境中的数据源同步上云的需求。
可以新增自定义的任务资源运行数据同步任务，解决 DataWorks 默认资源组与您的数据源不通的问题，或实现更高速度的传输能力。
当默认任务资源无法与您的复杂的网络环境连通时，可以通过数据集成自定义资源的部署，打通任意网络环境之间的数据传输同步。
2、自定义资源组的配置
(1)进入 Dataworks 控制台，点击需要数据同步的项目空间，点击数据集成。
(2)进入数据源界面，点击新增自定义资源组。
(3)确认 Kafka 与需要添加自定义资源组属于同一个 VPC 下。
(4)登录 ECS，执行命令 dmidecode|grep UUID 得到 ECS的 UUID。
(5)将 ECS 的 UUID 以及 IP，所占用资源的 CPU 与内存填写进来。
(6)在 ECS 上执行安装 Agent 的命令，添加完成测试连通性。
3、独享资源组的使用背景
独享资源模式下，机器的物理资源(网络、磁盘、CPU 和内存等)完全独享。
不仅可以隔离用户间的资源使用，也可以隔离不同工作空间任务的资源使用。此外，独享资源也支持灵活的扩容、缩容功能，可以满足资源独享、灵活配置等需求。
独享资源组可以访问同一地域的 VPC 数据源，也可以访问跨地域的公网 RDS 地址。
4、独享资源组的配置
(1)进入 DataWorks 控制台的资源组列表，点击新增独享集成资源组，点击购买选择对应的地区，CPU 以及内存。
(2)点击专有网路绑定，选择与 Kafka 对应 VPC 以及交换机(明显的区别是可用区)，安全组。
三、同步过程及其注意事项
1、DataWorks 数据集成操作
(1)进入 DataWorks 操作界面，点击创建业务流程，在新建的业务流程里添加数据同步节点。
(2)进入数据同步节点，点击数据源为 Kafka，点击转化为脚本模式。
2、Kafka Reader 的主要参数讲解
server：Kafka 的 broker server 地址，格式为 ip:poxt，是必填的。
topic：Kafka 的 topic，是 Kafka 处理资源的消息源(feeds of messages)的聚合，是必填的。
column：需要读取的 Kafka 数据，支持常量列、数据列和属性列。
常量列：使用单引号包裹的列为常量列，例如["abc"，"123"]。
数据列
如果您的数据是一个 JSON，支持获取 JSON 的属性，例如["event_id"].
如果您的数据是一个 JSON，支持获取 JSON 的嵌套子属性，例如["tag.desc"]。
属性列
_key_表示消息的 key。
_value_表示消息的完整内容。
_partition_表示当前消息所在分区。
_headers_表示当前消息 headers 信息。
_offset_表示当前消息的偏移量。
_timestamp_表示当前消息的时间戮。
完整示例如下：
"column"：[
"_key_"，
"_value_"，
"_partition_"，
"_offset_"，
"_timestamp_"，
"123",
"event_id"，
"tag. desc"
]
keyType ：Kafka 的 key 的类型，包括 BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG 和 SHORT，是必填的。
valueType：Kafka 的 value 的类型，包括 BYTEARRAY、DOUBLE、FLOAT、INTEGER、LONG 和 SHORT，是必填的。
beginDateTime：数据消费的开始时间位点，为时间范围(左闭右开)的左边界。yyyymmddhhmmss 格式的时间字符串，可以和时间属性配合使用。Kafka 0.10.2以上的版本支持此功能。需要和 beginOffset 二选一。(说明：beginDateTime 和 endDateTime 配合使用。)
endDateTime：数据消费的结束时间位点，为时间范围(左闭右开)的右边界。yyyymmddhhmmss 格式的时间字符串，可以和时间属性配合使用。Kafka 0.10.2以上的版本支持此功能。需要和 endOffset 二选一。(说明：endDateTime 和 beginDateTime 配合使用。)
beginOffset：数据消费的开始时间位点，您可以配置以下形式。
例如15553274的数字形式，表示开始消费的点位。
seekToBeginning：表示从开始点位消费数据。
seekToLast：表示从上次的偏移位置读取数据。
seekToEnd：表示从最后点位消费数据，会读取到空数据。
需要和 beginDateTime 二选一。
endOffset：数据消费的结束位点，用来控制什么时候应该结束数据消费任务退出，需要和 endDateTime 二选一。
skipExceedRecord：
Kafka 使用 public ConsumerRecords<K,V>poll (final Duration timeout)消费数据，一次 poll 调用获取的数据可能在 endOffset 或者 endDateTime 之外。skipExceedRecord 用来控制这些多余的数据是否写出到目的端。由于消费数据使用了自动点位提交，
建议：
Kafka0.10.2之前版本：建议 skipExceedRecord 配置为 false。
Kafka0.10.2及以上版本：建议 sKipExceedRecord 配置为 true。
不必填写，默认值为 false。
partition：Kafka 的一个 topic 有多个分区(parttion)，正常情况下数据同步任务是读取 topic(多个分区)一个点位区间的数据。您也可以指定 partition，仅读取一个分区点位区间的数据。不必填写，无默认值。
kafkaConfig：创建 Kafka 数据消费套户端 KafkaConsumer 可以指定扩展参数，例如 bootstrap-servers、auto.commit.interval.ms、Session.timeout.ms等，您可以基于 kafkaConfig控制KafkaConsumer 消费数据的行为，不必填写。
3、MaxCompute Writer 的主要参数讲解
datasource：数据源名称，脚本模式支持添加数据源，此配置项填写的内容必须与添加的数据源名称保持一致，是必填的。
table：写入的数据表的表名称(大小写不敏感)，不支持填写多张表，是必填的。
partition：需要写入数据表的分区信息，必须指定到最后一级分区。例如把数据写入一个三级分区表，必须配置到最后一级分区，例如 pt=20150101，type=1，biz=2。
·对于非分区表，该值务必不要填写，表示直接导入至目标表。
·MaxCompute Writer 不支持数据路由写入，对于分区表请务必保证写入数据到最后一级分区。
如果表为分区表，则必填。如果表为非分区表，则不能填写。
column：需要导入的字段列表。当导入全部字段时，可以配置为“column”:[“*”]。当需要插入部分 MaxCompute 列，则填写部分列，例如“column:["id","name"]。
·MaxCompute Writer 支持列筛选、列换序。例如一张表中有 a、b 和 c 三个字段，您只同步 c 和 b 两个字段，则可以配置为“column”:[“c”，“b”]，在导入过程中，字段 a 自动补空，设置为 null。
·column 必须显示指定同步的列集台，不允许为空。
是必填的。
truncate：通过配置“truncate”:“true”保证写入的幂等性。即当出现写入失败再次运行时，MaxCompute Writer 将清理前述数据，并导入新数据，可以保证每次重跑之后的数据都保持一致。
因为利用 MaxCompute SQL 进行数据清理工作，SQL 无法做到原子性，所以 truncate 选项不是原子操作。因此当多个任务同时向一个 Table/Partition 清理分区时，可能出现并发时序问题，请务必注意。
针对这类问题，建议您尽量不要多个作业 DDL 同时操作同一个分区，或者在多个井发作业启动前，提前创建分区。
是必填的。
4、Kafka2MaxCompute 脚本模式编写
{
"type"："job"，
"steps"：[{
"stepType"："kafka"，
"parameter"：{
"server"：“******"
"endOffset"："1000000"，
"kafkaConfig"：{
"group.id"：
"FinancesGroup"
}，
"valueType"："ByteArray"，
"column”：[
"__value__"，
"__timestamp__"，
"__partition__"，
"__offset__"
]，
"topic"： "finances_topic"，
"keyType"："ByteArray"，
"beginOffset"："seekToLast"
}，
"name":："Reader"，
"category"："reader"
}，
{
"stepType"： "odps"，
"parameter"：{
"truncate"：false，
"compress"：false，
"datasource"："odps_first"，
"column"：[
"value"，
"timestamp1"，
"partition"，
"offset"
]，
"emptyAsNull"：false，
"table"："testkafka3"
}，
"name"："Writer"，
"category"："writer"
}
]，
"version"："2.0"，
"order"：{
"hops"：[{
"from"："Reader"，
"to"："Writer"
}]
}，
"setting”：{
"errorLimit"：{
"record"："4"
}，
"speed”：{
"throttle"：false，
"concurrent"：1
}
}
}
5、Kafka 同步数据到 MaxCompute
(1)Kafka 的 Reader
{
"type"："job"，
"steps"：[{
"stepType"："kafka"，
"parameter"：{
"server"：“*******"
"endOffset"："1000000"，
"kafkaConfig"：{
"group.id"：
"FinancesGroup"
}，
"valueType"："ByteArray"，
"column”：[
"__value__"，
"__timestamp__"，
"__partition__"，
"__offset__"
]，
"topic"： "finances_topic"，
"keyType"："ByteArray"，
"beginOffset"："seekToLast"
}，
"name":："Reader"，
"category"："reader"
}，
(2) MaxCompute 的 Writer
{
"stepType"： "odps"，
"parameter"：{
"truncate"：false，
"compress"：false，
"datasource"："odps_first"，
"column"：[
"value"，
"timestamp1"，
"partition"，
"offset"
]，
"emptyAsNull"：false，
"table"："testkafka3"
}，
"name"："Writer"，
"category"："writer"
}
]，
(3)限制参数
"version"："2.0"，
"order"：{
"hops"：[{
"from"："Reader"，
"to"："Writer"
}]
}，
"setting”：{
"errorLimit"：{
"record"："4"
}，
"speed”：{
"throttle"：false，
"concurrent"：1
}
}
}
6、参考 Kafka 生产者 SDK 编写代码
示例代码如下：
//加载 kafka.properties
Properties kafkaProperties = JavaKafkaConfigurer.getKafka Properties(): Properties props = new Properties():
//设置接入点，即控制台的实例详情页显示的“默认接入点”
props.put(PreducerConfig.BOOTSTRAP_SERVERS_CONFIG,kafkaProperties.getProperty("bootatrap. servers”)):
//接入协议
props.put(CommonClientConfig =.SECURITY_PROTOCOL_CONFIG,“PLAINTEXT”):
// Kafka 消息的序列化方式
props.put(ProducerConfig.KHY_SERIALIZER_CLASS_CONFIG, "org, spache, kafka. common, serialination. StringSerializen”
props.put(ProducerConfig.VALUE_SERIAL.IZHR_CLASS_COMFIG,"ore.apache.kafka.common. gerislization.StringSerializen”
//请求的最长等待时间
props,put (ProducerConfig.MAX_DLOCK_MS_CONFIG，30 * 1000):
//构造 Producer 对象，注意，该对象是线程安全的，一般来说，一个进程内一个 Producer对象即可；
//如果想提高性能，可以多构造几个对象，但不要太多，最好不要超过5个
KafkaProducer<String, String>producer = nev KafkaProducer <String,String>(props):
//构造一个 Kafka 消息
String topic = kafkaProperties.getProperty(“topic”)://消息所属的 Topic，请在控制台申请之后，填写在这里
String value = "this is the message's value”://消息的内容
ProducerRecord<String, String> kafkaMessage = new ProducerR ecord<String, String>(topic, value);
Try {
//发送消息，并获得一个 Future 对象
Future<RecordMetadata> metadataFuture = producer.send(kafka Message):
//同步获得 Future 对象的结果
RecordMetadata recordMetadata = netadataFuture.get():
Syztem.out.println(“Produce ok:" + recordMetadata. toString ()):
] catch(Exceptien e) [
//要考虑重试
Syztem.out.println("error accurred”):
e.printStackTrace():
]
详细代码参考文档涉及到配置文件，消息来源，生产者消费者的代码模板：
https://help.aliyun.com/document detail/99957.html?spm =a2c4g.11186623.6.566.45fc54eayX69 b0
7、代码打包运行在在 ECS 上(与 Kafka 同一个可用区)
(1)执行 crontab-e 执行定时任务发送消息
017***java -jar/home/export/upload/javaCode3.jar >>/home/export/upload/logfile.log
(2)查看发送消息的定时任务日志
8、在 MaxCompute 上创建表
(1)创建目标表界面
(2)DDL 语句
CREATE TABLE `testkafka3`(`value` string,`timestamp1` string,` partition` string, `offset string)；
四、开发测试以及生产部署
1、选择自定义资源组(或独享集成资源组)进行同步操作
(1)选择可使用的独享资源组与自定义资源组进行同步。
(2)同步任务成功会显示，同步数据记录以及结果标志。
2、查询同步的数据结果
在 DataWorks 的临时界面查看同步数据结果。
3、设置调度参数
(1)点击右侧调度配置，输入调入时间。
(2)参考 DataWork 官方文档完善业务处理流程。
4、提交业务流程节点，并打包发布
(1)点击业务流程，提交业务流程节点。
(2)进入任务发布界面，将节点添加到待发布进行任务的部署。
5、确认业务流程发布成功
在运维中心页面，确认发布是否在生产环境中存在，至此 Kafka 同步数据到 MaxCompute 过程结束。
