elasticsearch-river-kafka 插件的环境配置和使用

1.elasticsearch-river-kafka 插件的安装

Github地址：https://github.com/endgameinc/elasticsearch-river-kafka

elasticsearch-river-kafka 插件的安装与其他插件一样

cd $ELASTICSEARCH_HOME

./bin/plugin -url file:/$PLUGIN_PATH -install elasticsearch-river-kafka

 

插件更新

cd $ELASTICSEARCH_HOME

./bin/plugin -remove elasticsearch-river-kafka

./bin/plugin -url file:/$PLUGIN_PATH -install elasticsearch-river-kafka

 

2.river节点的配置

配置river节点的时候，river节点和非river节点都要配置。

river节点：在es的配置文件中添加下面几行

#node.river: _none_    ##这一行要注释掉，表示为river节点

threadpool:

    bulk:

        type: fixed

        size: 60

        queue_size: 1000

 

非river节点：在es的配置文件中添加下面几行

node.river: _none_    ##这一行要解注，表示该节点不是river节点

threadpool:

    bulk:

        type: fixed

        size: 60

        queue_size: 1000

注意：一般，不会将数据落在river节点上（即node.data: false），但测试环境上就无所谓了，机器资源又紧张。

          节点配置完后，记得重启es，重启es的顺序：master节点→data节点→river节点

 

3.elasticsearch-river-kafka 插件的开发

社区中的elasticsearch-river-kafka 插件仅提供了对String和json数据的简单处理。在实际生产中，我们遇到的情况要复杂得多。

那么这个时候，我们就得自己去开发elasticsearch-river-kafka 插件实现一些附加功能。

下面就简单介绍一下开发elasticsearch-river-kafka 插件的步骤

1）KafkaRiverPlugin

该类需要继承KafkaRiverPlugin和实现AbstractPlugin，在该类中定义plugin的名称和描述

@Override

    public String name() {

        return "river-kafka";

    }

    @Override

    public String description() {

        return "River Kafka Plugin";

    }

 

2）es-plugin.properties配置文件

需要在es-plugin.properties中添加如下的定义，这样ES在启动的时候就能够通过org.elasticsearch.plugins.PluginManager

在当前的classpath中扫描到我们的plugin。

注意：定义中要写KafkaRiverPlugin类的全称，es-plugin.properties一般位于src/main/resources下

plugin=com.test.elasticsearch.plugin.river.kafka.KafkaRiverPlugin

 

3）KafkaRiverModule

KafkaRiverPlugin的onModule方法：在ES加载所有的插件时，会invoke一个onModule方法。KafkaRiverModule会作为参数传进来

public void onModule(RiversModule module) {

        module.registerRiver("kafka", KafkaRiverModule.class);

 }

KafkaRiverModule必须继承AbstractModule 。在KafkaRiverModule中会生成一个KafkaRiver。KafkaRiver是River接口的实现。

public class KafkaRiverModule extends AbstractModule {

    @Override

    protected void configure() {

        bind(River.class).to(KafkaRiver.class).asEagerSingleton();

    }

}

 

4）KafkaRiver

    – KafkaRiver必须继承AbstractRiverComponent，并且实现River接口。

    – KafkaRiver只提供两个方法：start和close。

    – AbstractRiverComponent 用于initialize kafkariver的logger、river名、river的配置

    – 构造函数通过@Inject注入river所需要的一切东西：RiverName, RiverSettings、logger、自定义的配置信息

      （这里是BasicProperties，在BasicProperties中定义的配置参数可以在创建river的时候被指定，参见“4.kafka→river→es的数据存储”）

    – 在start方法中启动了kafkariver的线程。在这个线程中，将数据从kafka中读取数据，然后将这些数据写到es中。

    – kafkaConsumer用来定义从kafka中读取数据时的用户操作。

    – ElasticsearchProducer用来定义将数据写入ES时的用户操作。

public class KafkaRiver extends AbstractRiverComponent implements River {

    private BasicProperties properties;

    private KafkaConsumer kafkaConsumer;

    private ElasticsearchProducer elasticsearchProducer;

    private static  ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();

    private Thread riverMonitorThread;

    private KafkaRiverSubMonitor kafkaRiverSubMonitor;

    private Thread thread;

    private ESLogger logger;

    @Inject

    protected KafkaRiver(RiverName riverName, RiverSettings settings, Client client) {

        super(riverName, settings);

        this.logger = Loggers.getLogger(getClass(), settings.globalSettings(), riverName);

        properties = new BasicProperties(settings);

        elasticsearchProducer = new ElasticsearchProducer(client, properties);

        kafkaConsumer = new KafkaConsumer(riverName, properties, elasticsearchProducer);

    }

    @Override

 

    public void start() {

     //启动KafkaRiver的线程

        try {

            logger.info("MHA: Starting Kafka Worker...");

            thread = EsExecutors.daemonThreadFactory(settings.globalSettings(), "kafka_river").newThread(kafkaConsumer);

            thread.start();

        } catch (Exception ex) {

            logger.error("Unexpected Error occurred", ex);

            throw new RuntimeException(ex);

        }

    }

     ......

}

 

4.kafka→river→es的数据存储

通过下面的指令，可以创建一条river，这样从kafka的baymaxtest的topic中的数据通过river就会落到es上。

注意：一个集群可以创建多个river，各river可以指定不同的topic、patition和序列化类

curl -XPUT 'http://localhost:9200/_river/baymaxriver1/_meta' -d '{

    "type": "kafka",

    "kafka": {

        "topic" : "test",

        "numOfConsumer" : "2",

        "zk.connect" : "10.10.10.10:2181",

        "zk.session.timeout.ms" : "50000",

        "zk.sync.time.ms" : "200",

        "zk.auto.commit.interval.ms" : "1000",

        "zk.auto.commit.enable" : "true",

        "zk.auto.offset.reset" : "smallest",

        "zk.fetch.message.max.bytes" : "5242880",

        "serializer" : "com.test.elasticsearch.river.kafka.serializer.AASerializer"

    },

     "elasticsearch" : {

        "indexName" : "stringfortest",

        "indexType" : "message1",

        "batch_size" : "500",

        "handling_batch_coresize" : "2",

        "handling_batch_maximumPoolSize" : "2",

        "handling_batch_keepAliveTime" : "600",

        "handling_batch_queueSize" : "10",

        "es_bulk_timeout" : "5"

    }

}'

上述指令中主要配置信息的说明：

kafka中  →

topic：kafka的topic名为test，

numOfConsumer：从kafka中读取数据的消费者个数

zk.connect：zookper的host名

serializer：对从kafka中来的数据的序列化类

elasticsearch中  →

indexName：在es中生成的index名，从该river中通过的数据会落到这个index中

indexType：index的type

es_bulk_timeout：es批量处理的timeout

上述指令会返回下面的结果

{"_index":"_river",

 "_type":"baymaxriver1",

 "_id":"_meta",

 "_version":1,

 "created":true

}

 

查看river的元数据：http://ip:9200/_river/rivername/_meta

 

删除一条river

curl -XDELETE 'http://localhost:9200/_river/rivername'