安裝

需要JDK 1.7以上版本。

直接解壓就可以使用了。

加入全域性path目錄：

export PATH=<path-to-confluent>/bin:$PATH

啟動

$ confluent start
Starting zookeeper
zookeeper is [UP]
Starting kafka
kafka is [UP]
Starting schema-registry
schema-registry is [UP]
Starting kafka-rest
kafka-rest is [UP]
Starting connect
connect
 
 is [UP]

檢視日誌檔案目錄

$ confluent current
# /tmp/confluent.BnzjkBY7

列出連線

$ confluent list connectors
Bundled Predefined Connectors (edit configuration under etc/):
  elasticsearch-sink
  file-source
  file-sink
  jdbc-source
  jdbc-sink
  hdfs-sink
  s3-sink

把本地檔案當作生產者源

屬性檔案在./etc/kafka/connect-file-source.properties

# User defined connector instance name.
name=file-source
# The class implementing the connector
connector.class=FileStreamSource
# Maximum number of tasks to run for this connector instance
tasks.max=1
# The input file (path relative to worker's working directory)
# This is the only setting specific to the FileStreamSource
 

file=test.txt
# The output topic in Kafka
topic=connect-test

弄一些測試資料：

$ for i in {1..3}; do echo "log line $i"; done > test.txt

載入檔案源

$ confluent load file-source
{
  "name": "file-source",
  "config": {
    "connector.class": "FileStreamSource",
    "tasks.max": "1",
    "file": "test.txt",
    "topics": "connect-test",
    "name": "file-source"
  },
  "tasks": []
}

檢視載入的聯結器

$ confluent status connectors
[
  "file-source"
]

檢視具體聯結器狀態

可以使用這個方法來排錯，經常使用。

$ confluent status file-source
{
  "name": "file-source",
  "connector": {
    "state": "RUNNING",
    "worker_id": "192.168.10.1:8083"
  },
  "tasks": [
    {
      "state": "RUNNING",
      "id": 0,
      "worker_id": "192.168.10.1:8083"
    }
  ]
}

在命令列裡檢視訊息

$ kafka-avro-console-consumer --bootstrap-server localhost:9092 --topic connect-test --from-beginning
  "log line 1"
  "log line 2"
  "log line 3"

使用聯結器寫檔案資料

使用聯結器file-sink，配置程式碼在./etc/kafka/connect-file-sink.properties:

# User defined name for the connector instance
name=file-sink
# Name of the connector class to be run
connector.class=FileStreamSink
# Max number of tasks to spawn for this connector instance
tasks.max=1
# Output file name relative to worker's current working directory
# This is the only property specific to the FileStreamSink connector
file=test.sink.txt
# Comma separate input topic list
topics=connect-test

載入消費檔案聯結器

$ confluent load file-sink
{
  "name": "file-sink",
  "config": {
    "connector.class": "FileStreamSink",
    "tasks.max": "1",
    "file": "test.sink.txt",
    "topics": "connect-test",
    "name": "file-sink"
  },
  "tasks": []
}

檢視狀態

$ confluent status file-sink
{
  "name": "file-sink",
  "connector": {
    "state": "RUNNING",
    "worker_id": "192.168.10.1:8083"
  },
  "tasks": [
    {
      "state": "RUNNING",
      "id": 0,
      "worker_id": "192.168.10.1:8083"
    }
  ]
}

現在就有倆聯結器了：

$ confluent status connectors
[
  "file-source",
  "file-sink"
]

檢視效果

$ tail -f test.sink.txt

隨機寫入一些資料：

$ for i in {4..1000}; do echo "log line $i"; done >> test.txt

出來了：

"log line 1"
"log line 2"
"log line 3"
"log line 4"
"log line 5"
 ...

解除安裝聯結器

$ confluent unload file-source
$ confluent unload file-sink

停用聯結器

$ confluent stop connect
Stopping connect
connect is [DOWN]

關閉confluent

$ confluent stop
Stopping connect
connect is [DOWN]
Stopping kafka-rest
kafka-rest is [DOWN]
Stopping schema-registry
schema-registry is [DOWN]
Stopping kafka
kafka is [DOWN]
Stopping zookeeper
zookeeper is [DOWN]

消滅痕跡

$ confluent destroy
Stopping connect
connect is [DOWN]
Stopping kafka-rest
kafka-rest is [DOWN]
Stopping schema-registry
schema-registry is [DOWN]
Stopping kafka
kafka is [DOWN]
Stopping zookeeper
zookeeper is [DOWN]
Deleting: /tmp/confluent.BnzjkBY7

參考