Kafka connect快速构建数据ETL通道

By | 2019年1月30日

一.背景介绍

Kafka connect是Confluent公司(当时开发出Apache Kafka的核心团队成员出来创立的新公司)开发的confluent platform的核心功能.

大家都知道现在数据的ETL过程经常会选择kafka作为消息中间件应用在离线和实时的使用场景中,而kafka的数据上游和下游一直没有一个

无缝衔接的pipeline来实现统一,比如会选择flume或者logstash采集数据到kafka,然后kafka又通过其他方式pull或者push数据到目标存储.

而kafka connect旨在围绕kafka构建一个可伸缩的,可靠的数据流通道,通过kafka connect可以快速实现大量数据进出kafka从而和其

他源数据源或者目标数据源进行交互构造一个低延迟的数据pipeline.给个图更直观点,大家感受下.

0

二.Kafka-connect快速配置

这里Confluent官方很贴心的提供了一个集成的镜像以便quickstart,如下链接

https://s3-us-west-2.amazonaws.com/confluent-files/kafka_connect_blog.ova

这是存储在Amazon S3上的,直接点击即可下载.这里我使用VMWare直接打开,刚开始会提示一个错误,不用管它直接点击重试即可

系统加载的过程中会默认初始化虚拟机的网络配置,这里我建议提前设置好桥接网络,让该虚拟机使用桥接网络初始化.

加载成功后,登录进入该Ubuntu系统,默认的用户名和密码都是:vagrant.

然后ls查看vagrant用户目录,查看几个关键的脚本内容后,我分别介绍它们的功能

1>setup.sh:自动下载mysql,mysql jdbc driver,配置好mysql以及做为hive的metastore

2>start.sh:启动confluent platform,kafka,hadoop,hive相关服务

3>clean_up.sh:和start.sh相反的,会关闭掉所有的服务,而且还会删除掉所有的数据(例如hdfs namenode和 datanode的数据,其实相当于fs format了)

那么很明显,第一步肯定是执行setup.sh,这里执行后会报错如下

setupFailed

这里无法下载相关的软件包,好吧,那么我们需要更新一下下载源的索引,执行如下命令

sudo apt-get update

更新完毕后再次执行setup.sh安装好mysql,hive等服务

紧接着执行start.sh来启动上述服务,启动后应该有如下进程,这是一个伪分布式节点

jpsService

对了,虚拟机各个服务(例如hive,zookeeper等),配置文件和日志文件在路径/mnt/下,组件的安装位置位于/opt下

三.Kafka connect快速使用

配置完以后就可以准备使用kafka-connect来快速构建一个数据pipeline了,如下图所示

wholePic

整个过程是将数据以mysql作为数据源,将数据通过kafka connect快速ETL到hive中去.注意这里图中没画kafka

但是实际上是包含在kafka connect里面的,话不多说,开始使用

1>Mysql数据准备

执行如下命令

复制代码

复制代码

注意上面第一行,–password=”mypassword” ,对,你没看错,这里虚拟机mysql的root默认密码就是mypassword,

强迫症患者请自行更改.随后建库,建表,插入数据.

2>关键概念准备

这里我快速普及一下参考官方文档理解的一些关键概念.

kafka connector:kafka connector是kafka connect的关键组成部分,它是一个逻辑上的job,用于在kafka和其他系统之间拷贝数据,比如

从上游系统拷贝数据到kafka,或者从kafka拷贝数据到下游系统

Tasks:每个kafka connector可以初始化一组task进行数据的拷贝

Workers:逻辑上包含kafka connector和tasks用来调度执行具体任务的进程,具体执行时分为standalone模式和distributed模式

见下图,这个是kafka上游的数据stream过来后,定义好对应的kafka connector后,分解为一组tasks然后push数据到kafka的不同topic

kafkaConnectors

3>利用Kafka-connect摄取数据

主要是通过配置来实现从mysql摄取数据到kafka,然后按照topic来获取数据写入hdfs,命令如下

注意上面这些properties文件是虚拟机已经事先配置好的,可以直接执行实现数据的摄取

当前使用的kafka connect的standalone模式,当然还有distributed模式后续可以尝试

上面的那条命令的格式是这样:

主要解释一下connect-standalone后面的参数

worker.properties:就是上面提到过的worker进程的配置文件,可以定义kafka cluster的相关信息以及数据序列化的格式.

随后的一些参数就是kafka connector的配置参数了,比如上面的mysql.properties定义了一个kafka jdbc connector,用来同步mysql数据到kafka

最后一个hdfs.properties是kafka hdfs connector的配置文件,用来消费kafka topic数据push到hdfs.

那么执行这条命令后就可以将mysql的数据通过kafka connect快速ETL到hdfs了.

最后可以通过hive创建外表映射hdfs上的数据文件,然后在hive中查看对应数据,如下

复制代码

复制代码

四.Kafka connect使用总结

1>Kafka connect的使用其实就是配置不同的kafka connectors,这里大家可以把kafka作为中间组件,然后可以类比flume理解,kafka上游的

connector其实就是fllume的source从上游数据源sink到kafka,kafka的下游connector其实就是flume的source是kafka,sink到下游系统.

2>Kafka connect的数据pipeline要打通,它要求数据遵守confluent自己的一套通用的schema机制,细心的同学会发现上面jps后会有个进程名

SchemaRegistryMain,这里官方默认使用Avro格式进出Kafka,所以要留意worker.properties文件的配置信息.

3>我在使用中没有发现Flume 相关的connector,因此很好奇它应该是没有实现上游flume conector的属性配置。问题应该出在Flume的数据是基

于event的,而和上面2中所说的schema定义格式没有很好的兼容.

4>kafka connect的distributed模式应该更实用,随后会尝试,以及confluent所支持的实时处理流kafka streams.

发表评论

电子邮件地址不会被公开。 必填项已用*标注