当前位置:   article > 正文

Kafka Connect简介_kafka connect sourcerecord keyschema

kafka connect sourcerecord keyschema

一. Kafka Connect简介

  Kafka是一个使用越来越广的消息系统,尤其是在大数据开发中(实时数据处理和分析)。为何集成其他系统和解耦应用,经常使用Producer来发送消息到Broker,并使用Consumer来消费Broker中的消息。Kafka Connect是到0.9版本才提供的并极大的简化了其他系统与Kafka的集成。Kafka Connect运用用户快速定义并实现各种Connector(File,Jdbc,Hdfs等),这些功能让大批量数据导入/导出Kafka很方便。

             

如图中所示,左侧的Sources负责从其他异构系统中读取数据并导入到Kafka中;右侧的Sinks是把Kafka中的数据写入到其他的系统中。

二. 各种Kafka Connector

  Kafka Connector很多,包括开源和商业版本的。如下列表中是常用的开源Connector

ConnectorsReferences
JdbcSourceSink
Elastic SearchSink1Sink2Sink3
CassandraSource1Source 2Sink1Sink2 
MongoDBSource
HBaseSink
SyslogSource
MQTT (Source)Source
Twitter (Source)SourceSink
S3Sink1Sink2
 

  商业版的可以通过Confluent.io获得

三. 示例

3.1 FileConnector Demo

 本例演示如何使用Kafka Connect把Source(test.txt)转为流数据再写入到Destination(test.sink.txt)中。如下图所示:

          

      本例使用到了两个Connector:

  • FileStreamSource:从test.txt中读取并发布到Broker中
  • FileStreamSink:从Broker中读取数据并写入到test.sink.txt文件中

  其中的Source使用到的配置文件是${KAFKA_HOME}/config/connect-file-source.properties

name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=test.txt
topic=connect-test

  其中的Sink使用到的配置文件是${KAFKA_HOME}/config/connect-file-sink.properties

name=local-file-sink
connector.class=FileStreamSink
tasks.max=1
file=test.sink.txt
topics=connect-test

  Broker使用到的配置文件是${KAFKA_HOME}/config/connect-standalone.properties

复制代码

bootstrap.servers=localhost:9092key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=trueinternal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000

复制代码

 

3.2 运行Demo

  需要熟悉Kafka的一些命令行,参考本系列之前的文章Apache Kafka系列(二) 命令行工具(CLI)

 3.2.1 启动Kafka Broker

[root@localhost bin]# cd /opt/kafka_2.11-0.11.0.0/
[root@localhost kafka_2.11-0.11.0.0]# ls
bin  config  libs  LICENSE  logs  NOTICE  site-docs
[root@localhost kafka_2.11-0.11.0.0]# ./bin/zookeeper-server-start.sh ./config/zookeeper.properties &
[root@localhost kafka_2.11-0.11.0.0]# ./bin/kafka-server-start.sh ./config/server.properties &

3.2.2 启动Source Connector和Sink Connector

[root@localhost kafka_2.11-0.11.0.0]# ./bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties 

3.3.3 打开console-consumer

./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic connect-test

3.3.4 写入到test.txt文件中,并观察3.3.3中的变化

[root@Server4 kafka_2.12-0.11.0.0]# echo 'firest line' >> test.txt
[root@Server4 kafka_2.12-0.11.0.0]# echo 'second line' >> test.txt
3.3.3中打开的窗口输出如下
{"schema":{"type":"string","optional":false},"payload":"firest line"}
{"schema":{"type":"string","optional":false},"payload":"second line"}

3.3.5 查看test.sink.txt

[root@Server4 kafka_2.12-0.11.0.0]# cat test.sink.txt 
firest line
second line

 本例仅仅演示了Kafka自带的File Connector,后续文章会完成JndiConnector,HdfsConnector,并且会使用CDC(Changed Data Capture)集成Kafka来完成一个ETL的例子

 

四. kafka 0.9 connect JDBC测试

kafka 0.9的connect功能,测试过程如下:

1.创建容器(本次采用docker容器构建kafka环境)

docker run -p 10924:9092 -p 21814:2181 --name confluent -i -t -d java /bin/bash

2.将confluent安装程序拷贝进容器;

docker cp  confluent.zip confluent:/root

3.进入到confluent容器

docker exec -it confluent /bin/bash

4.解压confluent压缩包

unzip confluent.zip

5.启动kafka

/root/confluent/bin/zookeeper-server-start  /root/confluent/etc/kafka/zookeeper.properties  & > zookeeper.log

/root/confluent/bin/kafka-server-start  /root/confluent/etc/kafka/server.properties & > server.log

/root/confluent/bin/schema-registry-start  /root/confluent/etc/schema-registry/schema-registry.properties & > schema.log

6.测试kafka 是否正常

开两个docker窗口,一个跑producer,一个跑consumer,

/root/confluent/bin/kafka-avro-console-producer --broker-list localhost:9092 --topic test --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}'

 

/root/confluent/bin/kafka-avro-console-consumer --topic test  --zookeeper localhost:2181 --from-beginning

在producer端依次输入以下记录,确认consumer能正确显示;

{"f1": "value1"}

{"f1": "value2"}

{"f1": "value3"}

以上为安装kafka过程,接下来开始测试jdbc接口;

测试之前,需要获取mysql JDBC的驱动并将获放在kafka环境对应的jre/lib文件夹里

测试jdbc connect

1.创建配置文件quickstart-mysql.properties,内容如下: 

name=test-mysql-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://192.168.99.100:33061/test1?user=root&password=welcome1
mode=incrementing
incrementing.column.name=id
topic.prefix=test-mysql-jdbc-

注:mysql是我在另一个容器里运行的,jdbc:mysql://192.168.99.100:33061/test1?user=root&password=welcome1是连接容器里的mysql的连接串

2.执行./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/quickstart-mysql.properties

3.执行./bin/kafka-avro-console-consumer --new-consumer --bootstrap-server 192.168.99.100:10924 --topic test-mysql-jdbc-accounts --from-beginning

然后在数据库里增加一条记录

然后就会在consumer端显示新增记录

 

五. 配置连接器

 

Connector的配置是简单的key-value映射。对于独立模式,这些都是在属性文件中定义,并通过在命令行上的Connect处理。在分布式模式,JSON负载connector的创建(或修改)请求。大多数配置都是依赖的connector,有几个常见的选项:

  • name - 连接器唯一的名称,不能重复。
  • connector.calss - 连接器的Java类。
  • tasks.max - 连接器创建任务的最大数。
  • connector.class配置支持多种格式:全名或连接器类的别名。比如连接器是org.apache.kafka.connect.file.FileStreamSinkConnector,你可以指定全名,也可以使用FileStreamSinkFileStreamSinkConnector。Sink connector也有一个额外的选项来控制它们的输入:
  • topics - 作为连接器的输入的topic列表。

对于其他的选项,你可以查看连接器的文档。

六、rest api

kafka connect的目的是作为一个服务运行,默认情况下,此服务运行于端口8083。它支持rest管理,用来获取 Kafka Connect 状态,管理 Kafka Connect 配置,Kafka Connect 集群内部通信,常用命令如下:

GET /connectors 返回一个活动的connect列表
POST /connectors 创建一个新的connect;请求体是一个JSON对象包含一个名称字段和连接器配置参数

GET /connectors/{name} 获取有关特定连接器的信息
GET /connectors/{name}/config 获得特定连接器的配置参数
PUT /connectors/{name}/config 更新特定连接器的配置参数
GET /connectors/{name}/tasks 获得正在运行的一个连接器的任务的列表

DELETE /connectors/{name} 删除一个连接器,停止所有任务,并删除它的配置

GET /connectors 返回一个活动的connect列表

POST /connectors 创建一个新的connect;请求体是一个JSON对象包含一个名称字段和连接器配置参数

GET /connectors/{name} 获取有关特定连接器的信息
GET /connectors/{name}/config 获得特定连接器的配置参数
PUT /connectors/{name}/config 更新特定连接器的配置参数
GET /connectors/{name}/tasks 获得正在运行的一个连接器的任务的列表

DELETE /connectors/{name} 删除一个连接器,停止所有任务,并删除它的配置

curl -s <Kafka Connect Worker URL>:8083/ | jq   获取 Connect Worker 信息

curl -s <Kafka Connect Worker URL>:8083/connector-plugins | jq 列出 Connect Worker 上所有 Connector

curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/tasks | jq 获取 Connector 上 Task 以及相关配置的信息

curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/status | jq 获取 Connector 状态信息

curl -s <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/config | jq 获取 Connector 配置信息

curl -s -X PUT <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/pause 暂停 Connector

curl -s -X PUT <Kafka Connect Worker URL>:8083/connectors/<Connector名字>/resume 重启 Connector

curl -s -X DELETE <Kafka Connect Worker URL>:8083/connectors/<Connector名字> 删除 Connector

创建新 Connector (以FileStreamSourceConnector举例)

 
  1. curl -s -X POST -H "Content-Type: application/json" --data

  2. '{"name": "<Connector名字>",

  3. "config":

  4. {"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",

  5. "key.converter.schemas.enable":"true",

  6. "file":"demo-file.txt",

  7. "tasks.max":"1",

  8. "value.converter.schemas.enable":"true",

  9. "name":"file-stream-demo-distributed",

  10. "topic":"demo-distributed",

  11. "value.converter":"org.apache.kafka.connect.json.JsonConverter",

  12. "key.converter":"org.apache.kafka.connect.json.JsonConverter"}

  13. }'

  14. http://<Kafka Connect Worker URL>:8083/connectors | jq

 

更新 Connector配置 (以FileStreamSourceConnector举例)

 
  1. curl -s -X PUT -H "Content-Type: application/json" --data

  2. '{"connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",

  3. "key.converter.schemas.enable":"true",

  4. "file":"demo-file.txt",

  5. "tasks.max":"2",

  6. "value.converter.schemas.enable":"true",

  7. "name":"file-stream-demo-distributed",

  8. "topic":"demo-2-distributed",

  9. "value.converter":"org.apache.kafka.connect.json.JsonConverter",

  10. "key.converter":"org.apache.kafka.connect.json.JsonConverter"}'

  11. <Kafka Connect Worker URL>:8083/connectors/file-stream-demo-distributed/config | jq

七、kafka connect + debezium,解析binlog至kafka

在已知kafka connect和debezium作用,会使用kafka的基础上,学会使用debezium来读取binlog,并通过kafka connect将读取的内容放入kafka topic中。 

基于kafka0.10.0和Debezium0.6,mysql5.6

kafka connect

  • Kafka Connect是一种用于Kafka和其他数据系统之间进行数据传输的工具。
  • 仅关注数据的复制,并且不处理其他任务
  • Kafka connect有两个概念,一个source,另一个是sink。source是把数据从一个系统拷贝到kafka里,sink是从kafka拷贝到另一个系统里。
  • 可使用插件,获取不同系统的数据。例如通过Debezium插件解析mysql的日志,获取数据。
  • 支持集群,可以通过REST API管理Kafka Connect。
  • 对数据的传输进行管理和监控。

Debezium

  • Debezium是一个分布式平台,可将现有数据库转换为事件流,因此应用程序可以立即查看并立即响应数据库中每一行的更改。
  • Debezium建立在Apache Kafka之上,并提供用于监视特定数据库管理系统的Kafka Connect兼容连接器。
  • 本教程使用Debezium监控binlog。

准备操作

  • mysql需开启binlog
 
  1. [mysqld]

  2. log-bin=mysql-bin #添加这一行就ok

  3. binlog-format=ROW #选择row模式

  4. server_id=1 #配置mysql replaction需要定义,不能和canal的slaveId重复

  • mysql需创建一个有mysql slave相关权限的账号,若mysql不在本机,则需要远程权限,防火墙放行。
 
  1. //mysql slave相关权限

  2. CREATE USER canal IDENTIFIED BY 'debezium';

  3. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';

  4. -- GRANT ALL PRIVILEGES ON *.* TO 'debezium'@'%' ;

  5. FLUSH PRIVILEGES;

  • 操作概述
  • 安装并启动kafka
  • 安装并启动mysql
  • 下载Debezium的mysql连接器http://debezium.io/docs/install/并解压
  • 安装debezium,即将解压目录写入classpath变量,例如:export classpath=/root/debezium-connector-mysql/* 
    只在当前shell有效
  • 参考http://debezium.io/docs/connectors/mysql/的配置文件示例,写好配置文件。
  • 以独立模式启动kafka connect,此时debezium会对数据库中的每一个表创建一个topic,消费相应的topic,即可获取binlog解析信息。
 
  1. //启动kafka connect

  2. bin/connect-standalone.sh config/connect-standalone.properties mysql.properties

  3. //查看topic列表

  4. bin/kafka-topics.sh --list --zookeeper localhost:2181

  5. //消费该主题

  6. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

  • 配置文件
 
  1. //mysql.properties

  2. name=inventory-connector

  3. connector.class=io.debezium.connector.mysql.MySqlConnector

  4. database.hostname=192.168.99.100

  5. database.port=3306

  6. database.user=debezium

  7. database.password=dbz

  8. database.server.id=184054

  9. database.server.name=fullfillment

  10. database.whitelist=inventory

  11. database.history.kafka.bootstrap.servers=192.168.30.30:9092

  12. database.history.kafka.topic=dbhistory.fullfillment

  13. include.schema.changes=true

  • 索引

debezium官网 http://debezium.io/ 

kafka文档 http://kafka.apache.org/0100/documentation.html

 

八、Kafka Connect的优点

1.对开发者提供了统一的实现接口
2.开发,部署和管理都非常方便,统一 
3.使用分布式模式进行水平扩展,毫无压力
4.在分布式模式下可以通过Rest Api提交和管理Connectors
5.对offset自动管理,只需要很简单的配置,而不像Consumer中需要开发者处理
6.流式/批式处理的支持

九、第三方资源

这是已经得到支持的组件,不需要做额外的开发: https://www.confluent.io/product/connectors/
括号中的Source表示将数据从其他系统导入Kafka,Sink表示将数据从Kafka导出到其他系统。
其他的我没看,但是JDBC的实现比较的坑爹,是通过primary key(如id)和时间戳(如updateTime)字段,

来判断数据是否更新,这样的话应用范围非常受局限。

 

十、Connector Development Guide

 

在kafka与其他系统间复制数据需要创建kafka connect,他们将数复制到kafka或者从kafka复制到其他系统

连接器有两种形式:sourceconnectors将另一个系统数据导入kafka,sinkconnectors将数据导出到另一个系统

连接器不执行任何数据复制:它们的描述复制的数据,并且负责将工作分配给多个task

task分为sourcetask与sinktask

每个task从kafka复制数据,connect会保证record与schema的一致性完成任务分配,通常record与schema的映射是明显的,每一个文件对应一个流,流中的每一条记录利用schema解析并且保存对应的offset,另外一种情况是我们需要自己完成这种映射,比如数据库,表的offset不是很明确(没有自增id),一种可能的选择是利用时间(timestamp)来完成增量查询。

Streams and Records


每一个stream是包含key value对的记录的序列,key value可以是原始类型,可以支持复杂结构,除了array,object,嵌套等。数据转换是框架来完成的,record中包含stream id与offset,用于定时offset提交,帮助当处理失败时恢复避免重复处理。

Dynamic Connectors

所有的job不是静态的,它需要监听外部系统的变化,比如数据库表的增加删除,当一个新table创建时,它必须发现并且更新配置由框架来分配给该表一个task去处理,当通知发布后框架会更新对应的task.

Developing a Simple Connector

例子很简单
在standalone模式下实现 SourceConnector/SourceTask 读取文件并且发布record给SinkConnector/SinkTask 由sink写入文件

Connector Example

我们将实现SourceConnector,SinkConnector实现与它非常类似,它包括两个私有字段存放配置信息(读取的文件名与topic名称)
public class FileStreamSourceConnector extends SourceConnector {
    private String filename;
    private String topic;
getTaskClass()方法定义实现执行处理的task
@Override
public Class getTaskClass() {
    return FileStreamSourceTask.class;
}
下面定义FileStreamSourceTask,它包括两个生命周期方法start,stop
@Override
public void start(Map<String, String> props) {
    // The complete version includes error handling as well.
    filename = props.get(FILE_CONFIG);
    topic = props.get(TOPIC_CONFIG);
}
@Override
public void stop() {
    // Nothing to do since no background monitoring is required.
}
最后是真正核心的方法getTaskConfigs()在这里我们仅处理一个文件,所以我们虽然定义了max task(在配置文件里)但是只会返回一个包含一条entry的list
@Override
public List<Map<String, String>> getTaskConfigs(int maxTasks) {
    ArrayList>Map<String, String>> configs = new ArrayList<>();
    // Only one input stream makes sense.
    Map<String, String> config = new Map<>();
    if (filename != null)
        config.put(FILE_CONFIG, filename);
    config.put(TOPIC_CONFIG, topic);
    configs.add(config);
    return configs;
}
即使有多个任务,这种方法的执行通常很简单。它只是要确定输入任务的数量,这可能需要拉取数据从远程服务,然后分摊。请注意,这个简单的例子不包括动态输入。在下一节中看到讨论如何触发任务的配置更新。

Task Example - Source Task

实现task,我们使用伪代码描述核心代码
public class FileStreamSourceTask extends SourceTask<Object, Object> {
    String filename;
    InputStream stream;
    String topic;
    public void start(Map<String, String> props) {
        filename = props.get(FileStreamSourceConnector.FILE_CONFIG);
        stream = openOrThrowError(filename);
        topic = props.get(FileStreamSourceConnector.TOPIC_CONFIG);
    }
    @Override
    public synchronized void stop() {
        stream.close()
    }
start方法读取之前的offset,并且处理新的数据,stop方法停止stream,下面实现poll方法
@Override
public List<SourceRecord> poll() throws InterruptedException {
    try {
        ArrayList<SourceRecord> records = new ArrayList<>();
        while (streamValid(stream) && records.isEmpty()) {
            LineAndOffset line = readToNextLine(stream);
            if (line != null) {
                Map sourcePartition = Collections.singletonMap("filename", filename);
                Map sourceOffset = Collections.singletonMap("position", streamOffset);
                records.add(new SourceRecord(sourcePartition, sourceOffset, topic, Schema.STRING_SCHEMA, line));
            } else {
                Thread.sleep(1);
            }
        }
        return records;
    } catch (IOException e) {
        // Underlying stream was killed, probably as a result of calling stop. Allow to return
        // null, and driving thread will handle any shutdown if necessary.
    }
    return null;
}
该方法重复执行读取操作,跟踪file offset,并且利用上述信息创建SourceRecord,它需要四个字段:source partition,source offset,topic name,output value(包括value及value的schema)

Sink Tasks

之前描述了sourcetask实现,sinktask与它完全不同,因为前者是拉取数据,后者是推送数据
public abstract class SinkTask implements Task {
public void initialize(SinkTaskContext context) { ... }
public abstract void put(Collection<SinkRecord> records);
public abstract void flush(Map<TopicPartition, Long> offsets);
put方法是最重要的方法,接收sinkrecords,执行任何需要的转换,并将其存储在目标系统。此方法不需要确保数据已被完全写入目标系统,然后返回。事实上首先放入缓冲,因此,批量数据可以被一次发送,减少对下游存储的压力。sourcerecords中保存的信息与sourcesink中的相同。flush提交offset,它接受任务从故障中恢复,没有数据丢失。该方法将数据推送至目标系统,并且block直到写入已被确认。的offsets参数通常可以忽略不计,但在某些情况保存偏移信息到目标系统确保一次交货。例如,一个HDFS连接器可以确保flush()操作自动提交数据和偏移到HDFS中的位置。

Resuming from Previous Offsets

kafka connect是为了bulk 数据拷贝工作,它拷贝整个db而不是拷贝某个表,这样会使用connnect的input或者output随时改变,source connector需要监听source系统的改变,当改变时通知框架(通过ConnectorContext对象)
举例
if (inputsChanged())
    this.context.requestTaskReconfiguration();
当接收到通知框架会即时的更新配置,并且在更新前确保优雅完成当前任务
如果一个额外的线程来执行此监控,该线程必须存在于连接器中。该线程不会影响connector。然而,其他变化也会影响task,最常见的是输入流失败在输入系统中,例如如果一个表被从数据库中删除。这时连接器需要进行更改,任务将需要处理这种异常。sinkconnectors只能处理流的加入,可以分配新的数据到task(例如,一个新的数据库表)。框架会处理任何kafka输入的改变,例如当组输入topic的变化因为一个正则表达式的订阅。sinktasks应该期待新的输入流,可能需要在下游系统创造新的资源,如数据库中的一个新的表。在这些情况下,可能会出现输入流之间的冲突(同时创建新资源),其他时候,一般不需要特殊的代码处理一系列动态流  

Dynamic Input/Output Streams

FileStream连接器是很好的例子,因为他们很简单的,每一行是一个字符串。实际连接器都需要具有更复杂的数据格式。要创建更复杂的数据,你需要使用kafka connector数据接口:Schema,Struct
Schema schema = SchemaBuilder.struct().name(NAME)
                    .field("name", Schema.STRING_SCHEMA)
                    .field("age", Schema.INT_SCHEMA)
                    .field("admin", new SchemaBuilder.boolean().defaultValue(false).build())
                    .build();
Struct struct = new Struct(schema)
                           .put("name", "Barbara Liskov")
                           .put("age", 75)
                           .build();
如果上游数据与schema数据格式不一致应该在sinktask中抛出异常

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/你好赵伟/article/detail/782238
推荐阅读
相关标签
  

闽ICP备14008679号