kafka connect

红太狼 2022-02-20 11:30 180阅读 0赞

### 一. Kafka Connect简介 ###

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

![314515-20170821220746418-613069141.png][]

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

### 二. 各种Kafka Connector ###

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

<table> 
 <tbody> 
  <tr> 
   <td>Connectors</td> 
   <td>References</td> 
  </tr> 
  <tr> 
   <td>Jdbc</td> 
   <td><a href="https://github.com/apache/ignite/tree/master/modules/kafka" rel="nofollow">Source</a>,&nbsp;<a href="https://github.com/apache/ignite/tree/master/modules/kafka" rel="nofollow">Sink</a></td> 
  </tr> 
  <tr> 
   <td>Elastic Search</td> 
   <td><a href="https://github.com/ksenji/kafka-connect-es" rel="nofollow">Sink1</a>,&nbsp;<a href="https://github.com/hannesstockner/kafka-connect-elasticsearch" rel="nofollow">Sink2</a>,&nbsp;<a href="https://github.com/DataReply/kafka-connect-elastic-search-sink" rel="nofollow">Sink3</a></td> 
  </tr> 
  <tr> 
   <td>Cassandra</td> 
   <td><a href="https://github.com/tuplejump/kafka-connect-cassandra" rel="nofollow">Source1</a>,&nbsp;<a href="https://github.com/datamountaineer/stream-reactor/tree/master/kafka-connect-cassandra" rel="nofollow">Source 2</a>,&nbsp;<a href="https://github.com/tuplejump/kafka-connect-cassandra" rel="nofollow">Sink1</a>,&nbsp;<a href="https://github.com/datamountaineer/stream-reactor/tree/master/kafka-connect-cassandra" rel="nofollow">Sink2&nbsp;</a></td> 
  </tr> 
  <tr> 
   <td>MongoDB</td> 
   <td><a href="https://github.com/DataReply/kafka-connect-mongodb" rel="nofollow">Source</a></td> 
  </tr> 
  <tr> 
   <td>HBase</td> 
   <td><a href="https://github.com/mravi/kafka-connect-hbase" rel="nofollow">Sink</a></td> 
  </tr> 
  <tr> 
   <td>Syslog</td> 
   <td><a href="https://github.com/jcustenborder/kafka-connect-syslog" rel="nofollow">Source</a></td> 
  </tr> 
  <tr> 
   <td>MQTT (Source)</td> 
   <td><a href="https://github.com/evokly/kafka-connect-mqtt" rel="nofollow">Source</a></td> 
  </tr> 
  <tr> 
   <td>Twitter (Source)</td> 
   <td><a href="https://github.com/rollulus/twitter-kafka-connect" rel="nofollow">Source</a>,&nbsp;<a href="https://github.com/Eneco/kafka-connect-twitter" rel="nofollow">Sink</a></td> 
  </tr> 
  <tr> 
   <td>S3</td> 
   <td><a href="https://github.com/qubole/streamx" rel="nofollow">Sink1</a>,&nbsp;<a href="https://github.com/DeviantArt/kafka-connect-s3" rel="nofollow">Sink2</a><br> &nbsp;</td> 
  </tr> 
 </tbody> 
</table>

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

### 三. 示例 ###

3.1 FileConnector Demo

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

![314515-20170821225909980-1168791908.png][]

本例使用到了两个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

![复制代码][copycode.gif]

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

![复制代码][copycode.gif]

3.2 运行Demo

需要熟悉Kafka的一些命令行，参考本系列之前的文章[Apache Kafka系列(二) 命令行工具（CLI）][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`，你可以指定全名，也可以使用`FileStreamSink`或`FileStreamSinkConnector`。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对象包含一个名称字段和连接器配置参数

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/][http_debezium.io_docs_install]并解压
 *  安装debezium，即将解压目录写入classpath变量，例如：export classpath=/root/debezium-connector-mysql/\*  
    只在当前shell有效
 *  参考[http://debezium.io/docs/connectors/mysql/][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/][http_debezium.io]

kafka文档 [http://kafka.apache.org/0100/documentation.html][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/][Confluent.io]  
括号中的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][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中抛出异常

## 参考 ##

[Kafka Documentation][]

http://kafka.apache.org/documentation/\#connect

https://www.confluent.io/product/connectors/

[314515-20170821220746418-613069141.png]: /images/20220220/dc00cddef07a428db74b37c3706995f1.png
[Confluent.io]: https://www.confluent.io/product/connectors/
[314515-20170821225909980-1168791908.png]: /images/20220220/117c146f7fc84814b11838749d6b4ccc.png
[copycode.gif]: /images/20220220/7d6a0fb8b4074999bf910c4becd70795.png
[Apache Kafka_ _CLI]: http://www.cnblogs.com/qizhelongdeyang/p/7354315.html
[http_debezium.io_docs_install]: http://debezium.io/docs/install/
[http_debezium.io_docs_connectors_mysql]: http://debezium.io/docs/connectors/mysql/
[http_debezium.io]: http://debezium.io/
[http_kafka.apache.org_0100_documentation.html]: http://kafka.apache.org/0100/documentation.html
[Streams and Records]: http://kafka.apache.org/documentation.html#connect_streamsandrecords
[Dynamic Connectors]: http://kafka.apache.org/documentation.html#connect_dynamicconnectors
[Developing a Simple Connector]: http://kafka.apache.org/documentation.html#connect_developing
[Connector Example]: http://kafka.apache.org/documentation.html#connect_connectorexample
[Task Example - Source Task]: http://kafka.apache.org/documentation.html#connect_taskexample
[Sink Tasks]: http://kafka.apache.org/documentation.html#connect_sinktasks
[Resuming from Previous Offsets]: http://kafka.apache.org/documentation.html#connect_resuming
[Dynamic Input_Output Streams]: http://kafka.apache.org/documentation.html#connect_dynamicio
[Kafka Documentation]: https://kafka.apache.org/documentation/#connect