Kafka数据迁移MaxCompute最佳实践

本文涉及的产品
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 本文向您详细介绍如何使用DataWorks数据同步功能,将Kafka集群上的数据迁移到阿里云MaxCompute大数据计算服务。

前提条件
搭建Kafka集群
进行数据迁移前,您需要保证自己的Kafka集群环境正常。本文使用阿里云EMR服务自动化搭建Kafka集群,详细过程请参见:Kafka 快速入门。

本文使用的EMR Kafka版本信息如下:
EMR版本: EMR-3.12.1
集群类型: Kafka
软件信息: Ganglia 3.7.2 ZooKeeper 3.4.12 Kafka 2.11-1.0.1 Kafka-Manager 1.3.3.16
Kafka集群使用专有网络,区域为华东1(杭州),主实例组ECS计算资源配置公网及内网IP,具体配置如下图所示。

创建MaxCompute 项目
开通MaxCompute服务并创建好项目,本文中在华东1(杭州)区域创建项目bigdata_DOC,同时启动DataWorks相关服务,如下图所示。详情请参见开通MaxCompute。

背景信息
Kafka是一款分布式发布与订阅的消息中间件,具有高性能、高吞量的特点被广泛使用,每秒能处理上百万的消息。Kafka适用于流式数据处理,主要应用于用户行为跟踪、日志收集等场景。

一个典型的Kafka集群包含若干个生产者(Producer)、Broker、消费者(Consumer)以及一个Zookeeper集群。Kafka集群通过Zookeeper管理自身集群的配置并进行服务协同。

Topic是Kafka集群上最常用的消息的集合,是一个消息存储逻辑概念。物理磁盘不存储Topic,而是将Topic中具体的消息按分区(Partition)存储在集群中各个节点的磁盘上。每个Topic可以有多个生产者向它发送消息,也可以有多个消费者向它拉取(消费)消息。

每个消息被添加到分区时,会分配一个offset(偏移量,从0开始编号),是消息在一个分区中的唯一编号。

操作步骤
准备测试表与数据
Kafka集群创建测试数据
为保证您可以顺利登陆EMR集群Header主机及MaxCompute和DataWorks可以顺利和EMR集群Header主机通信,请您首先配置EMR集群Header主机安全组,放通TCP 22及TCP 9092端口。
登录EMR集群Header主机地址
进入EMR Hadoop控制台集群管理 > 主机列表页面,确认EMR集群Header主机地址,并通过SSH连接远程登录。

创建测试Topic
使用kafka-topics.sh --zookeeper emr-header-1:2181/kafka-1.0.1 --partitions 10 --replication-factor 3 --topic testkafka --create命令创建测试所使用的Topic testkafka。您可以使用kafka-topics.sh --list --zookeeper emr-header-1:2181/kafka-1.0.1命令查看已创建的Topic。
[root@emr-header-1 ~]# kafka-topics.sh --zookeeper emr-header-1:2181/kafka-1.0.1 --partitions 10 --replication-factor 3 --topic testkafka --create
Created topic "testkafka".
[root@emr-header-1 ~]# kafka-topics.sh --list --zookeeper emr-header-1:2181/kafka-1.0.1
__consumer_offsets
_emr-client-metrics
_schemas
connect-configs
connect-offsets
connect-status
testkafka
写入测试数据
您可以使用kafka-console-producer.sh --broker-list emr-header-1:9092 --topic testkafka命令模拟生产者向Topic testkafka中写入数据。由于Kafka用于处理流式数据,您可以持续不断的向其中写入数据。 为保证测试结果,建议您写入10条以上的数据。
[root@emr-header-1 ~]# kafka-console-producer.sh --broker-list emr-header-1:9092 --topic testkafka

123
abc

为验证写入数据生效,您可以同时再打开一个SSH窗口,使用kafka-console-consumer.sh --bootstrap-server emr-header-1:9092 --topic testkafka --from-beginning命令模拟消费者,核验数据是否已成功写入Kafka。 如下所示,当数据写入成功时,您可以看到已写入的数据。

[root@emr-header-1 ~]# kafka-console-consumer.sh --bootstrap-server emr-header-1:9092 --topic testkafka --from-beginning
123
abc
创建MaxCompute表
为保证MaxCompute可以顺利接收Kafka数据,请您首先在MaxCompute上创建表。本例中为测试便利,使用非分区表。
登陆DataWorks创建表,详情请参见表管理。

您可以单击DDL模式进行建表,建表语句举例如下。
CREATE TABLE testkafka (

`key` string,
`value` string,
`partition1` string,
`timestamp1` string,
`offset` string,
`t123` string,
`event_id` string,
`tag` string

) ;
其中的每一列,对应于DataWorks数据集成Kafka Reader的默认列,您可以自主命名。详情请参见配置Kafka Reader:
__key__表示消息的key。
__value__表示消息的完整内容 。
__partition__表示当前消息所在分区。
__headers__表示当前消息headers信息。
__offset__表示当前消息的偏移量。
__timestamp__表示当前消息的时间戳。
数据同步
新建自定义资源组
由于当前DataWorks的默认资源组无法完美支持Kafka插件,您需要使用自定义资源组完成数据同步。自定义资源组详情请参见新增任务资源。

在本文中,为节省资源,直接使用EMR集群Header主机作为自定义资源组。完成后,请等待服务器状态变为可用。

新建并运行同步任务
在您的业务流程中右键单击数据集成,选择新建数据集成节点 > 数据同步。

新建数据同步节点后,您需要选择数据来源的数据源为Kafka,数据去向的数据源为ODPS,并且使用默认数据源odps_first。选择数据去向表为您新建的testkafka。完成上述配置后,请点击下图框中的按钮,转换为脚本模式。

脚本配置如下,代码释义请参见配置Kafka Reader。
{

"type": "job",
"steps": [
    {
        "stepType": "kafka",
        "parameter": {
            "server": "47.xxx.xxx.xxx:9092",
            "kafkaConfig": {
                "group.id": "console-consumer-83505"
            },
            "valueType": "ByteArray",
            "column": [
                "__key__",
                "__value__",
                "__partition__",
                "__timestamp__",
                "__offset__",
                "'123'",
                "event_id",
                "tag.desc"
            ],
            "topic": "testkafka",
            "keyType": "ByteArray",
            "waitTime": "10",
            "beginOffset": "0",
            "endOffset": "3"
        },
        "name": "Reader",
        "category": "reader"
    },
    {
        "stepType": "odps",
        "parameter": {
            "partition": "",
            "truncate": true,
            "compress": false,
            "datasource": "odps_first",
            "column": [
                "key",
                "value",
                "partition1",
                "timestamp1",
                "offset",
                "t123",
                "event_id",
                "tag"
            ],
            "emptyAsNull": false,
            "table": "testkafka"
        },
        "name": "Writer",
        "category": "writer"
    }
],
"version": "2.0",
"order": {
    "hops": [
        {
            "from": "Reader",
            "to": "Writer"
        }
    ]
},
"setting": {
    "errorLimit": {
        "record": ""
    },
    "speed": {
        "throttle": false,
        "concurrent": 1,
        "dmu": 1
    }
}

}
您可以通过在Header主机上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --list命令查看group.id参数,及消费者的Group名称。
[root@emr-header-1 ~]# kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --list
Note: This will not show information about old Zookeeper-based consumers.

_emr-client-metrics-handler-group
console-consumer-69493
console-consumer-83505
console-consumer-21030
console-consumer-45322
console-consumer-14773
以console-consumer-83505为例,您可以根据该参数在Header主机上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505命令确认beginOffset及endOffset参数。
[root@emr-header-1 ~]# kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505
Note: This will not show information about old Zookeeper-based consumers.
Consumer group 'console-consumer-83505' has no active members.
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
testkafka 6 0 0 0 - - -
test 6 3 3 0 - - -
testkafka 0 0 0 0 - - -
testkafka 1 1 1 0 - - -
testkafka 5 0 0 0 - - -
完成脚本配置后,请首先切换任务资源组为您刚创建的资源组,然后点击运行。

完成运行后,您可以在运行日志中查看运行结果,如下为成功运行的日志。

结果验证
您可以通过新建一个数据开发任务运行SQL语句,查看当前表中是否已存在从Kafka同步过来的数据。本例中使用select * from testkafka;语句,完成后点击运行即可。

执行结果如下,本例中为保证结果,在testkafka Topic中输入了多条数据,您可以查验是否和您输入的数据一致。

目录
相关文章
|
2月前
|
分布式计算 搜索推荐 物联网
大数据及AI典型场景实践问题之通过KafKa+OTS+MaxCompute完成物联网系统技术重构如何解决
大数据及AI典型场景实践问题之通过KafKa+OTS+MaxCompute完成物联网系统技术重构如何解决
|
2月前
|
消息中间件 数据采集 关系型数据库
大数据-业务数据采集-FlinkCDC 读取 MySQL 数据存入 Kafka
大数据-业务数据采集-FlinkCDC 读取 MySQL 数据存入 Kafka
42 1
|
3月前
|
消息中间件 分布式计算 NoSQL
EMR-Kafka Connect:高效数据迁移的革新实践与应用探索
Kafka Connect是Kafka官方提供的一个可扩展的数据传输框架,它允许用户以声明式的方式在Kafka与其他数据源之间进行数据迁移,无需编写复杂的数据传输代码。
|
3月前
|
消息中间件 存储 Kafka
如何通过 CloudCanal 实现从 Kafka 到 AutoMQ 的数据迁移
随着大数据技术的飞速发展,Apache Kafka 作为一种高吞吐量、低延迟的分布式消息系统,已经成为企业实时数据处理的核心组件。然而,随着业务的扩展和技术的发展,企业面临着不断增加的存储成本和运维复杂性问题。为了更好地优化系统性能和降低运营成本,企业开始寻找更具优势的消息系统解决方案。其中,AutoMQ [1] 作为一种基于云重新设计的消息系统,凭借其显著的成本优势和弹性能力,成为了企业的理想选择。
23 0
如何通过 CloudCanal 实现从 Kafka 到 AutoMQ 的数据迁移
|
3月前
|
存储 分布式计算 监控
日志数据投递到MaxCompute最佳实践
日志服务采集到日志后,有时需要将日志投递至MaxCompute的表中进行存储与分析。本文主要向用户介绍将数据投递到MaxCompute完整流程,方便用户快速实现数据投递至MaxCompute。
166 2
|
2月前
|
存储 设计模式 分布式计算
面向对象编程在大数据处理中的最佳实践
【8月更文第12天】随着互联网和物联网技术的发展,数据量呈指数级增长,大数据处理已成为现代企业不可或缺的一部分。大数据处理通常涉及收集、存储、管理和分析海量数据集。传统的数据库管理系统难以应对这样的挑战,因此出现了诸如Hadoop、Spark等分布式处理框架。这些框架通常使用面向对象编程(OOP)来构建可扩展、可维护的应用程序。本文将探讨如何利用面向对象编程的原则和模式来优化大数据处理任务。
63 0
|
3月前
|
消息中间件 分布式计算 大数据
大数据处理工具及其与 Kafka 的搭配使用
大数据处理工具及其与 Kafka 的搭配使用
48 2
|
4月前
|
消息中间件 大数据 Kafka
高效处理大数据:Kafka的13个核心概念详解
大家好,我是小米!今天我将为大家深入解析Kafka的核心概念,包括消息、批次、主题、分区、副本、生产者、消费者、消费组等内容。通过这篇文章,你将全面了解Kafka的工作机制和应用场景,为你的大数据处理提供有力支持。准备好了吗?让我们开始吧!
129 4
|
3月前
|
分布式计算 DataWorks 调度
MaxCompute产品使用合集之如何将数据迁移到CDH Hive
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
4月前
|
消息中间件 存储 大数据
深度分析:Apache Kafka及其在大数据处理中的应用
Apache Kafka是高吞吐、低延迟的分布式流处理平台,常用于实时数据流、日志收集和事件驱动架构。与RabbitMQ(吞吐量有限)、Pulsar(多租户支持但生态系统小)和Amazon Kinesis(托管服务,成本高)对比,Kafka在高吞吐和持久化上有优势。适用场景包括实时处理、数据集成、日志收集和消息传递。选型需考虑吞吐延迟、持久化、协议支持等因素,使用时注意资源配置、数据管理、监控及安全性。

相关产品

  • 云原生大数据计算服务 MaxCompute
  • 下一篇
    无影云桌面