背景

我們在做大數(shù)據(jù)的項(xiàng)目時(shí)經(jīng)常遇到數(shù)據(jù)同步的問題偷拔，kafka到es是一個(gè)比較常見的數(shù)據(jù)同步管道。有這樣的一個(gè)需求亏钩，kafka入es的過程中莲绰，我們需要達(dá)到如下效果：kafka中同一個(gè)topic數(shù)據(jù)流可以動(dòng)態(tài)的插入到es不同的type里面，當(dāng)然铸屉，它們是同一個(gè)index钉蒲。什么叫動(dòng)態(tài)插入切端？舉個(gè)例子：譬如我們按照時(shí)間來切分type彻坛，每個(gè)小時(shí)為單位，那么踏枣，在三點(diǎn)到四點(diǎn)之間昌屉，數(shù)據(jù)就會同步到type為2018011003里面，在四點(diǎn)到五點(diǎn)之間茵瀑，數(shù)據(jù)就會同步到type為2018011004里面间驮。將這個(gè)需求往深處想的話，同樣马昨，是不是可以實(shí)現(xiàn)動(dòng)態(tài)index插入呢竞帽？答案是可以的扛施。

實(shí)現(xiàn)原理

river這個(gè)es插件大家不知道聽過沒，我們可以利用river實(shí)現(xiàn)該功能屹篓。網(wǎng)上有很多現(xiàn)成的開源插件elasticsearch-river-kafka疙渣，它們都有個(gè)共同點(diǎn)就是只能靜態(tài)的同步數(shù)據(jù)。這里的靜態(tài)指的是不能動(dòng)態(tài)生成index和type堆巧。所以妄荔，我們基于開源庫elasticsearch-river-kafka，可以自己動(dòng)手修改代碼谍肤，來實(shí)現(xiàn)我們想要的功能啦租。而且，這個(gè)插件本身就是支持自定義開發(fā)的荒揣。

步驟

kafka：v0.10
elasticsearch：v1.7.0

一：代碼庫地址

github：elasticsearch-river-kafka
我是fork別人的開源項(xiàng)目篷角，在此基礎(chǔ)上自己修改的代碼，原項(xiàng)目地址

二：安裝該插件

• 方式一
  去你安裝elasticsearch的機(jī)器上乳附，找到es下的bin目錄執(zhí)行安裝插件命令

cd $ELASTICSEARCH_HOME
./bin/plugin --install <plugin-name> --url https://github.com/zhuyinglinfeng/elasticsearch-river-kafka/archive/master.zip

• 方式二
  如果你本地已經(jīng)clone了該項(xiàng)目内地，直接本地編譯打包，然后把zip包發(fā)送到安裝elasticsearch的機(jī)器上面赋除，再執(zhí)行安裝插件命令

cd $ELASTICSEARCH_HOME
./bin/plugin --install <plugin-name> --url file:<zip_path>

三：配置river（這個(gè)是重點(diǎn)哦）

一條river代表著一條同步規(guī)則阱缓，創(chuàng)建river很簡單，執(zhí)行如下的命令即可

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參數(shù)航背，表示index根據(jù)時(shí)間動(dòng)態(tài)創(chuàng)建，index參數(shù)可以不用填寫棱貌，即使填寫也不會生效
如果填寫了frequency.type參數(shù)玖媚，表示type根據(jù)時(shí)間動(dòng)態(tài)創(chuàng)建，type參數(shù)可以不用填寫婚脱，即使填寫也不會生效

Example:
將kafka中topic名字為topic-test流中的數(shù)據(jù)同步到index為index-test里面今魔，并且每隔10min動(dòng)態(tài)切換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

查詢進(jìn)程號：ps -ef | grep elastic
關(guān)閉進(jìn)程：kill -9 pid
啟動(dòng)：./bin/elasticsearch -d

五：驗(yàn)證

到此為止，理論上已經(jīng)全部完成障贸，現(xiàn)在可以往kafka中寫數(shù)據(jù)了错森，不出意外，如果你的<flush.interval>參數(shù)設(shè)置很短的話篮洁，幾秒后es中就可以查看數(shù)據(jù)了

升級維護(hù)

• 插件刪除

./bin/plugin --remove <plugin-name>

• 刪除river
  和刪除elasticsearch數(shù)據(jù)方式一樣涩维，利用es提供的接口命令刪除即可

curl -XDELETE 'localhost:9200/_river/river-test/'

• 代碼升級
  如果你想要手動(dòng)改寫代碼，也很簡單袁波，fork一下代碼庫瓦阐，自己本地升級代碼蜗侈，然后按照上面的步驟就可以

自定義elasticsearch-river-kafka開發(fā)

......待完善