一. 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

三. 示例

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

 3.2.1 啟動Kafka Broker

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

3.2.2 啟動Source Connector和Sink Connector

[[email protected] 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中的變化

[[email protected] kafka_2.12-0.11.0.0]# echo 'firest line' >> test.txt
[[email protected] 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

[[email protected] 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的例子

 PS:

相比編譯過Kafka-Manager都知道各種坑，經過了3個小時的努力，我終於把Kafka-Manager編譯通過並打包了，並且新增了Kafka0.11.0版本支援。

附下載地址： 連結: https://pan.baidu.com/s/1miiMsAk 密碼: 866q