2018/1/13 20:31:44当前位置推荐好文程序员浏览文章

背景

我们在做大数据的项目时经常遇到数据同步的问题,kafka到es是一个比较常见的数据同步管道。有这样的一个需求,kafka入es的过程中,我们需要达到如下效果:kafka中同一个topic数据流可以动态的插入到es不同的type里面,当然,它们是同一个index。什么叫动态插入?举个例子:譬如我们按照时间来切分type,每个小时为单位,那么,在三点到四点之间,数据就会同步到type为2018011003里面,在四点到五点之间,数据就会同步到type为2018011004里面。将这个需求往深处想的话,同样,是不是可以实现动态index插入呢?答案是可以的。

实现原理

river这个es插件大家不知道听过没,我们可以利用river实现该功能。网上有很多现成的开源插件elasticsearch-river-kafka,它们都有个共同点就是只能静态的同步数据。这里的静态指的是不能动态生成index和type。所以,我们基于开源库elasticsearch-river-kafka,可以自己动手修改代码,来实现我们想要的功能。而且,这个插件本身就是支持自定义开发的。

步骤

kafka:v0.10
elasticsearch:v1.7.0

一:代码库地址

github:elasticsearch-river-kafka
我是fork别人的开源项目,在此基础上自己修改的代码,原项目地址

二:安装该插件

  • 方式一
    去你安装elasticsearch的机器上,找到es下的bin目录执行安装插件命令
cd $ELASTICSEARCH_HOME./bin/plugin --install <plugin-name> --url  zhuyinglinfeng/elasticsearch-river-kafka/archive/master.zip
  • 方式二
    如果你本地已经clone了该项目,直接本地编译打包,然后把zip包发送到安装elasticsearch的机器上面,再执行安装插件命令
cd $ELASTICSEARCH_HOME./bin/plugin --install <plugin-name> --url file:<zip_path>

三:配置river(这个是重点哦)

一条river代表着一条同步规则,创建river很简单,执行如下的命令即可

curl -XPUT localhost:9200/_river/<river-name>/_meta -d {     "type" : "kafka",     "kafka" : {        "zookeeper.connect" : <zookeeper.connect>,         "zookeeper.connection.timeout.ms" : <zookeeper.connection.timeout.ms>,        "topic" : <topic.name>,        "message.type" : <message.type>    },    "index" : {        "index" : <index.name>,        "frequency.index" : <frequency.index>,        "type" : <mapping.type.name>,        "frequency.type" : <frequency.type>,        "bulk.size" : <bulk.size>,        "concurrent.requests" : <concurrent.requests>,        "action.type" : <action.type>,        "flush.interval" : <flush.interval>    },    "statsd" : {        "host" : <statsd.host>,        "prefix" : <statsd.prefix>,        "port" : <statsd.port>,        "log.interval" : <statsd.log.interval>    } }
参数名字是否必填默认值描述
river-name名字
zookeeper.connectlocalhostzoo的地址
zookeeper.connection.timeout.ms10000zoo连接超时时间
topicelasticsearch-river-kafkatopic名字
message.typejsonkafka消息类型,json/string
indexkafka-indexES索引
frequency.index动态索引切分频率,1mon/1day/1hour/10min
typestatusES类型
frequency.type动态类型切分频率,1mon/1day/1hour/10min
bulk.size100单次处理的消息数量
concurrent.requests1并发请求数
action.typeindex同步行为,index(插入)/delete(删除)/raw.execute(执行语句)
hostlocalhoststatsd服务地址
port8125statsd端口
prefixkafka-riverstatsd键值前缀
log.interval10statsd上报metrics时间间隔

Note:
如果填写了frequency.index参数,表示index根据时间动态创建,index参数可以不用填写,即使填写也不会生效
如果填写了frequency.type参数,表示type根据时间动态创建,type参数可以不用填写,即使填写也不会生效

Example:
将kafka中topic名字为topic-test流中的数据同步到index为index-test里面,并且每隔10min动态切换type类型

curl -XPUT localhost:9200/_river/river-test/_meta -d {    "type": "kafka",    "kafka": {        "zookeeper.connect": "localhost",        "zookeeper.connection.timeout.ms": 10000,        "topic": "topic-test",        "message.type": "json"    },    "index": {        "index": "index-test",        "frequency.type”: "10min",        "bulk.size": 100,        "concurrent.requests": 1,        "action.type": "index",        "flush.interval": "5s"    }}

四:重启elasticsearch

查询进程号:ps -ef | grep elastic
关闭进程:kill -9 pid
启动:./bin/elasticsearch -d

五:验证

到此为止,理论上已经全部完成,现在可以往kafka中写数据了,不出意外,如果你的<flush.interval>参数设置很短的话,几秒后es中就可以查看数据了

升级维护

  • 插件删除
./bin/plugin --remove <plugin-name>
  • 删除river
    和删除elasticsearch数据方式一样,利用es提供的接口命令删除即可
curl -Xlocalhost:9200/_river/river-test/
  • 代码升级
    如果你想要手动改写代码,也很简单,fork一下代码库,自己本地升级代码,然后按照上面的步骤就可以

自定义elasticsearch-river-kafka开发

......待完善

网友评论