基于腾讯云kafka同步到Elasticsearch初解方式有几种？

Apache Kafka brokers (plain text):9092
Confluent Control Center:9021
Kafka Connect REST API:8083
REST Proxy:8082
Schema Registry REST API:8081
ZooKeeper:2181

[root@kafka_no1 confluent-3.3.0]# pwd
/home/confluent/confluent-3.3.0
[root@kafka_no1 confluent-3.3.0]# ls -al
total 32
drwxrwxr-x. 7 root root 4096 Dec 16 10:08 .
drwxr-xr-x. 3 root root 4096 Dec 20 15:34 ..
drwxr-xr-x. 3 root root 4096 Jul 28 08:30 bin
drwxr-xr-x. 18 root root 4096 Jul 28 08:30 etc
drwxr-xr-x. 2 root root 4096 Dec 21 15:34 logs
-rw-rw-r--. 1 root root 871 Jul 28 08:45 README
drwxr-xr-x. 10 root root 4096 Jul 28 08:30 share
drwxrwxr-x. 2 root root 4096 Jul 28 08:45 src

$ confluent start 1

Starting zookeeper
zookeeper is [UP] ——对应端口：2181
Starting kafka
kafka is [UP]——对应端口：9092
Starting schema-registry
schema-registry is [UP]——对应端口：8081
Starting kafka-rest
kafka-rest is [UP]
Starting connect
connect is [UP]

./kafka-avro-console-producer  --broker-list 110.118.7.11:9092 --topic test-elasticsearch-sink \
         --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}'
{"f1": "value1"}
{"f1": "value2"}
{"f1": "value3"}

./kafka-avro-console-consumer --bootstrap-server 110.118.7.11:9092 :9092 --topic  test-elasticsearch-sink --from-beginning

./connect-standalone ../etc/schema-registry/connect-avro-standalone.properties \
../etc/kafka-connect-elasticsearch/quickstart-elasticsearch.properties
注意此处： connect-standalone模式，对应 connect-avro-standalone.properties要修改；
如果使用connect-distribute模式，对应的connect-avro-distribute.properties要修改。
这里 quickstart-elasticsearch.properties ：启动到目的Elasticsearch配置。
quickstart-elasticsearch.properties**设置**：
name=elasticsearch-sink
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
#kafka主题名称，也是对应Elasticsearch索引名称
topics= test-elasticsearch-sink
key.ignore=true
#ES url信息
connection.url=http://110.18.6.20:9200
#ES type.name固定
type.name=kafka-connect

-
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并删除配置。