|
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'
|
|
|
|