背景

(1) 接近实时的数据计算都基于kafka(spark【doing】、flink【todo】)且spark提供sql相关操作接口

(2) 不同类型数据库提供查询接口多、异；(之前尝试直接通过接口直接再不同存储间读写，不走kafka，痛苦)

(3) 知识背景：主线(mysql：SQL、索引、CAP)，多条辅线(各集群参数调优、锁机制等)

接触confluent，偶然意识：将数据先入kafka，通过sql进行处理，再写目标数据库，各种不同连接器，使得kafka很灵活，一定程度简化未知的庞大的知识脉络；

暂定的知识脉络

kafka-connect

Connectors -->Tasks --> Workers -- >Converters (抽象到具体)Connector：负责任务调度进行数据由源到目标的转换，Connector大体有两类：Source(其他db数据->kafka);Sink(数据从kafka到其他db)两类Task：一个Connector涉及多个Task协调进行工作Worker：具体的任务执行Converter：数据格式转化，confluent默认提供如下四种格式的数据转化：AvroConverter、JsonConverter、StringConverter、ByteArrayConverterConnect操作

实际操作涉及：配置文件修改+topic创建+启动connect

配置文件修改worker.properties(以connect-standalone.properties为例)bootstrap.servers=XXXX

key.converter=org.apache.kafka.connect.json.JsonConverter

value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=true

value.converter.schemas.enable=true

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

plugin.path = /usr/share/confluent-hub-components

plugin.path=/usr/share/java,/usr/share/confluent-hub-components

# 如下两个配置用于RestAPI进行链接器信息访问

rest.host.name=XXXX

rest.port=XXX

配置文件修改：source.properties(以mysql为例)name=XXXXconnector.class=io.confluent.connect.jdbc.JdbcSourceConnectortasks.max=2 #connector.class 不同数据库source、sink的类不同connection.url=jdbc:mysql://IP:端口/库名?user=用户名&password=密码mode=incrementingincrementing.column.name=id # 自增主键列名topic.prefix=test-mysql-jdbc- # kafka中topic前缀，拼接表名，构成真实topic配置文件修改：sink.properties(以es为例)name=XXXXconnector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnectortasks.max=1topics=test-elasticsearch-sink # kafka中的topic名字，对应ES的index名字key.ignore=trueconnection.url=http://IP:port # ES访问的IP、端口type.name=kafka-connect备注：如ES中的index不存在，自动会创建，无需定义mapping；topic创建：将配置文件中涉及到的topic提前进行创建启动connect：将涉及到的配置拼接到命令行后，下面展示standalone模式下的启动：connect-standalone worker.properties source.properties sink.properties

KSQL

hive, spark, flink都有sql支持入口，kafka也有sql，所以一定要掌握好sql