kafka connector 使用總結以及自定義connector開發

Kafaka connect 是一種用于在Kafka和其他系統之間可擴展的、可靠的流式傳輸資料的工具，它使得能夠快速定義將大量資料集合移入和移出Kafka的連接器變得簡單，Kafka Connect可以從資料庫或應用程式服務器收集資料到Kafka topic，使資料可用于低延遲的流處理，匯出作業可以將資料從Kafka topic傳輸到二次存盤和查詢系統，或者傳遞到批處理系統以進行離線分析，

Kafaka connect的核心組件：
Source：負責將外部資料寫入到kafka的topic中，
Sink：負責從kafka中讀取資料到自己需要的地方去，比如讀取到HDFS，hbase等，

Connectors ：通過管理任務來協調資料流的高級抽象
Tasks：資料寫入kafk和從kafka中讀出資料的具體實作，source和sink使用時都需要Task

Workers：運行connectors和tasks的行程

Converters：kafka connect和其他存盤系統直接發送或者接受資料之間轉換資料，

converter會把bytes資料轉換成kafka connect內部的格式，也可以把kafka connect內部存盤格式的資料轉變成bytes，converter對connector來說是解耦的，所以其他的connector都可以重用，例如，使用了avro converter，那么jdbc connector可以寫avro格式的資料到kafka，當然，hdfs connector也可以從kafka中讀出avro格式的資料，

Transforms：一種輕量級資料調整的工具
Kafka connect 作業模式：
Kafka connect 有兩種作業模式：
standalone：在standalone模式中，所有的worker都在一個獨立的行程中完成，
distributed：distributed模式具有高擴展性，以及提供自動容錯機制，你可以使用一個group.ip來啟動很多worker行程，在有效的worker行程中它們會自動的去協調執行connector和task，如果你新加了一個worker或者掛了一個worker，其他的worker會檢測到然后在重新分配connector和task，

本文作者：張永清，轉載請注明出處：https://www.cnblogs.com/laoqing/p/11927958.html 

在分布式模式下通過rest api來管理connector，
connector的常見管理操作API：

GET /connectors – 回傳所有正在運行的connector名，POST /connectors – 新建一個connector; 請求體必須是json格式并且需要包含name欄位和config欄位，name是connector的名字，config是json格式，必須包含你的connector的配置資訊，GET /connectors/{name} – 獲取指定connetor的資訊，GET /connectors/{name}/config – 獲取指定connector的配置資訊，PUT /connectors/{name}/config – 更新指定connector的配置資訊，GET /connectors/{name}/status – 獲取指定connector的狀態，包括它是否在運行、停止、或者失敗，如果發生錯誤，還會列出錯誤的具體資訊，GET /connectors/{name}/tasks – 獲取指定connector正在運行的task，GET /connectors/{name}/tasks/{taskid}/status – 獲取指定connector的task的狀態資訊，PUT /connectors/{name}/pause – 暫停connector和它的task，停止資料處理知道它被恢復，PUT /connectors/{name}/resume – 恢復一個被暫停的connector，POST /connectors/{name}/restart – 重啟一個connector，尤其是在一個connector運行失敗的情況下比較常用POST /connectors/{name}/tasks/{taskId}/restart – 重啟一個task，一般是因為它運行失敗才這樣做，DELETE /connectors/{name} – 洗掉一個connector，停止它的所有task并洗掉配置，

如何開發自己的Connector：

1、引入maven依賴，

        <dependency>            <groupId>org.apache.kafka</groupId>            <artifactId>connect-api</artifactId>            <version>${kafka.version}</version>        </dependency>

2、開發自定義的Source

開發自定義的Source 需要繼承實作SourceConnector和SourceTask這兩個抽象類，實作抽象類中的未實作的方法或者重寫抽象類中的方法，

本文作者：張永清，轉載請注明出處：https://www.cnblogs.com/laoqing/p/11927958.html 

A、開發自定義的SourceConnector

/** * */public class ExampleSourceConnector extends SourceConnector{    @Override    public void start(Map<String, String> map) {    }    //回傳需要指定的TASK    @Override    public Class<? extends Task> taskClass() {        return ExampleSourceTask.class;    }    //TASK的配置    @Override    public List<Map<String, String>> taskConfigs(int i) {        return null;    }    @Override    public void stop() {    }    @Override    public ConfigDef config() {        return null;    }    @Override    public String version() {        return AppInfoParser.getVersion();    }}

B、開發Source對應的Task

public class ExampleSourceTask extends SourceTask {    @Override    public String version() {        return new ExampleSourceConnector().version();    }    //任務啟動    @Override    public void start(Map<String, String> map) {    }    //需要發送到kafka的資料，    @Override    public List<SourceRecord> poll() throws InterruptedException {        return null;    }    //任務停止    @Override    public void stop() {    }}

3、開發自定義的Sink

　　開發自定義的Sink 需要繼承實作SinkConnector和SinkTask這兩個抽象類，實作抽象類中的未實作的方法或者重寫抽象類中的方法，

A、開發自定義的SinkConnector

/** * */public class ExampleSinkConnector extends SinkConnector{    @Override    public void start(Map<String, String> map) {    }    //指定Task執行的類    @Override    public Class<? extends Task> taskClass() {        return ExampleSinkTask.class;    }    //task對應的config    @Override    public List<Map<String, String>> taskConfigs(int i) {        return null;    }    @Override    public void stop() {    }    //配置定義    @Override    public ConfigDef config() {        return null;    }    @Override    public String version() {        return AppInfoParser.getVersion();    }}

B、開發Sink對應的Task　　

/** * */public class ExampleSinkTask extends SinkTask {    @Override    public String version() {        return new ExampleSinkConnector().version();    }    //task啟動    @Override    public void start(Map<String, String> map) {    }    //資料put    @Override    public void put(Collection<SinkRecord> collection) {    }    @Override    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets){    //Task停止，    }    @Override    public void stop() {    }}

Kafka Connect Configs

 開源的實作的比較好的connector專案：

https://github.com/debezium/debezium 

https://github.com/confluentinc

 https://docs.confluent.io/current/connect/managing/connectors.html

 這里我們以https://github.com/debezium/debezium 中的debezium-connector-mongodb 為例配置connector的standalone模式運行

從github中獲取debezium-connector-mongodb-0.9.5.Final.jar 包，放到kafka的libs目錄下，并且把mongodb相關的jar包一起放入到libs下，

在config目錄下新建對應的mongodb.properties 屬性組態檔

name=mongodbconnector.class=io.debezium.connector.mongodb.MongoDbConnectormongodb.hosts=configs/10.100.xx.xx:27017tasks.max=1mongodb.name=mongo-test#mongodb.user=root#mongodb.password=123456database.whitelist=kafkaTestcollection.whitelist=kafkaTest.kafkaTestconnect.max.attempts=12max.queue.size=8192max.batch.size=2048poll.interval.ms=1000connect.backoff.initial.delay.ms=1000connect.backoff.max.delay.ms=2000mongodb.ssl.enabled=falsemongodb.ssl.invalid.hostname.allowed=falsesnapshot.mode=initialinitial.sync.max.threads=2tombstones.on.delete=truemongodb.members.auto.discover=truesource.struct.version=v2

 配置解釋如下：

詳情參考：https://debezium.io/documentation/reference/0.10/connectors/mongodb.html

https://docs.confluent.io/current/connect/debezium-connect-mongodb/mongodb_source_connector_config.html

The following advanced configuration properties have good defaults that will work in most situations and therefore rarely need to be specified in the connector’s configuration.

這里以standalone的模式運行，在connect-standalone.properties中做如下配置：

# Licensed to the Apache Software Foundation (ASF) under one or more# contributor license agreements.  See the NOTICE file distributed with# this work for additional information regarding copyright ownership.# The ASF licenses this file to You under the Apache License, Version 2.0# (the "License"); you may not use this file except in compliance with# the License.  You may obtain a copy of the License at##    http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.# These are defaults. This file just demonstrates how to override some settings.bootstrap.servers=localhost:9092rest.port=9093# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will# need to configure these based on the format they want their data in when loaded from or stored into Kafkakey.converter=org.apache.kafka.connect.json.JsonConvertervalue.converter=org.apache.kafka.connect.json.JsonConverter# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply# it tokey.converter.schemas.enable=falsevalue.converter.schemas.enable=falserest.host.name=0.0.0.0offset.storage.file.filename=/data4/kafka/connect/connect.offsets# Flush much faster than normal, which is useful for testing/debuggingoffset.flush.interval.ms=10000# Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins# (connectors, converters, transformations). The list should consist of top level directories that include# any combination of:# a) directories immediately containing jars with plugins and their dependencies# b) uber-jars with plugins and their dependencies# c) directories immediately containing the package directory structure of classes of plugins and their dependencies# Note: symlinks will be followed to discover dependencies or plugins.# Examples:# plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,plugin.path=/opt/kafka/kafka_2.11-2.0.0/plugin

standalone模式下啟動方式如下：

bin/connect-standalone.sh config/connect-standalone.properties connector1.properties[connector2.properties ...]   一次可以啟動多個connector，只需要在引數中加上connector的組態檔路徑即可，

例如：connect-standalone.sh config/connect-standalone.properties mongodb.properties

distribute模式部署：

1、修改配置connect-distributed.properties

# broker串列bootstrap.servers=10.120.241.1:9200 # 同一集群中group.id需要配置一致，且不能和別的消費者同名group.id=connect-cluster # The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will# need to configure these based on the format they want their data in when loaded from or stored into Kafkakey.converter=org.apache.kafka.connect.json.JsonConvertervalue.converter=org.apache.kafka.connect.json.JsonConverter# 使用json資料同樣配置成falsekey.converter.schemas.enable=falsevalue.converter.schemas.enable=false····

2、手動創建集群模式所必須的kafka的幾個topic

# config.storage.topic=connect-configs$ bin/kafka-topics --create --zookeeper localhost:2181 --topic connect-configs --replication-factor 3 --partitions 1 --config cleanup.policy=compact # offset.storage.topic=connect-offsets$ bin/kafka-topics --create --zookeeper localhost:2181 --topic connect-offsets --replication-factor 3 --partitions 50 --config cleanup.policy=compact # status.storage.topic=connect-status$ $ bin/kafka-topics --create --zookeeper localhost:2181 --topic connect-status --replication-factor 3 --partitions 10 --config cleanup.policy=compact

• config.storage.topic：topic用于存盤connector和任務配置；注意，這應該是一個單個的partition，多副本的topic
• offset.storage.topic：用于存盤offsets；這個topic應該配置多個partition和副本，
• status.storage.topic：用于存盤狀態；這個topic 可以有多個partitions和副本

3、 啟動worker

啟動distributed模式命令如下：

./bin/connect-distributed ./etc/kafka/connect-distributed.properties   

4、使用restful啟動connect

curl 'http://localhost:8083/connectors' -X POST -i -H "Content-Type:application/json" -d       '{ "name":"elasticsearch-sink",         "config":{"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",                  "tasks.max":10,                  "topics":"estest1012",                  "key.ignore":true,                  "schema.ignore":true,                  "connection.url":"http://10.120.241.194:9200",                  "type.name":"kafka-connect"}      }'　

常見問題：

1、在啟動的程序中出現各種各樣的java.lang.ClassNotFoundException，

在啟動connector的時候，一開始總是會報各個各樣的ClassNotFoundException，不是這個包就是那個包，查找問題一直說要么缺少包要么是包沖突，那么要排除依賴沖突或者看下是不是少了jar包，

2、在connector.properties中的key.converter.schemas.enable=false和value.converter.schemas.enable=false的問題，

這個選項默認在connect-standalone.properties中是true的，這個時候發送給topic的Json格式是需要使用avro格式，例如：

{    "schema": {        "type": "struct",        "fields": [{            "type": "int32",            "optional": true,            "field": "c1"        }, {            "type": "string",            "optional": true,            "field": "c2"        }, {            "type": "int64",            "optional": false,            "name": "org.apache.kafka.connect.data.Timestamp",            "version": 1,            "field": "create_ts"        }, {            "type": "int64",            "optional": false,            "name": "org.apache.kafka.connect.data.Timestamp",            "version": 1,            "field": "update_ts"        }],        "optional": false,        "name": "foobar"    },    "payload": {        "c1": 10000,        "c2": "bar",        "create_ts": 1501834166000,        "update_ts": 1501834166000    }}

如果想發送普通的json格式而不是avro格式的話，很簡單key.converter.schemas.enable和value.converter.schemas.enable設定為false就行，這樣就能發送普通的json格式資料，

標籤：大數據

上一篇：阿里云ECS服務器部署HADOOP集群（六）：Flume 安裝

下一篇：資料中臺架構體系淺析