背景

我们在做大数据的项目时经常遇到数据同步的问题，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 https://github.com/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>
    }
 }'

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 -XDELETE 'localhost:9200/_river/river-test/'

• 代码升级
  如果你想要手动改写代码，也很简单，fork一下代码库，自己本地升级代码，然后按照上面的步骤就可以

自定义elasticsearch-river-kafka开发

......待完善