flume-ng 与elasticsearch整合

àì夳堔傛蜴生んèń 2022-07-12 08:09 119阅读 0赞

*  http://blog.csdn.net/u010022051/article/details/50515725

实现过程：

JDK版本：1.7.0\_79

Elasticsearch版本：2.1.1

Flume版本：1.6

在flume的配置文件目录下添加如下配置文件

vim es\_log.conf

agent.sources = tail

agent.channels = memoryChannel  
agent.channels.memoryChannel.type = memory  
  
agent.sources.tail.channels = memoryChannel  
agent.sources.tail.type = spooldir  
agent.sources.tail.spoolDir = /home/elk/es\_log.log  
agent.sources.tail.fileHeader = true  
  
com.frontier45.flume.sink.elasticsearch2.ElasticSearchSink  
agent.sinks = elasticsearch  
agent.sinks.elasticsearch.channel = memoryChannel  
agent.sinks.elasticsearch.type=org.apache.flume.sink.elasticsearch.ElasticSearchSink  
agent.sinks.elasticsearch.batchSize=100  
agent.sinks.elasticsearch.hostNames=172.26.40.74:9300,172.26.40.75:9300,172.26.40.76:9300,172.27.40.77:9300,172.28.40.78:9300  
agent.sinks.k1.indexType = bar\_type  
agent.sinks.elasticsearch.indexName=logstash  
agent.sinks.elasticsearch.clusterName=elk  
agent.sinks.elasticsearch.serializer=org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer  
com.frontier45.flume.sink.elasticsearch2.ElasticSearchDynamicSerializer

新建数据文件目录

在Flume安装目录下执行如下命令启动Flume

bin/flume-ng agent -c /home/elk/apache-flume-1.6.0-bin/conf -f /home/elk/apache-flume-1.6.0-bin/conf/es\_log.conf -n agent -Dflume.root.logger=INFO,console

错误一：  
2016-01-11 14:46:32,260 (conf-file-poller-0) \[ERROR - org.apache.flume.sink.elasticsearch.ElasticSearchSink.configure(ElasticSearchSink.[Java][]:302)\] Could not instantiate event serializer.  
java.lang.ClassNotFoundException: org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer  
at java[.NET][].URLClassLoader$1.run(URLClassLoader.java:366)  
at java[.Net][.NET].URLClassLoader$1.run(URLClassLoader.java:355)  
at java.security.AccessController.doPrivileged(Native Method)  
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)  
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)  
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)  
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)  
at java.lang.Class.forName0(Native Method)  
at java.lang.Class.forName(Class.java:191)  
at org.apache.flume.sink.elasticsearch.ElasticSearchSink.configure(ElasticSearchSink.java:286)  
at org.apache.flume.conf.Configurables.configure(Configurables.java:41)  
at org.apache.flume.node.AbstractConfigurationProvider.loadSinks(AbstractConfigurationProvider.java:413)  
at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:98)  
at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140)  
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)  
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)  
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)  
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)  
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)  
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)  
at java.lang.Thread.run(Thread.java:745)  
  
原因：  
缺少Elasticsearch中的依赖包  
  
  
解决方案：  
1. 将Elasticsearch中lib下的jar包导入到Flume的lib下  
2. 在环境变量中引入Elasticsearch的依赖：  
FLUME\_CLASSPATH="/home/elk/elasticsearch-2.1.1/lib/\*"

错误二：  
2016-01-11 14:52:29,899 (lifecycleSupervisor-1-3) \[ERROR - org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:253)\] Unable to start SinkRunner: \{ policy:org.apache.flume.sink.DefaultSinkProcessor@1c9f6ece counterGroup:\{ name:null counters:\{\} \} \} - Exception follows.  
java.lang.NoSuchMethodError: org.elasticsearch.common.transport.InetSocketTransportAddress.<init>(Ljava/lang/String;I)V  
at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.configureHostnames(ElasticSearchTransportClient.java:143)  
at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.<init>(ElasticSearchTransportClient.java:77)  
at org.apache.flume.sink.elasticsearch.client.ElasticSearchClientFactory.getClient(ElasticSearchClientFactory.java:48)  
at org.apache.flume.sink.elasticsearch.ElasticSearchSink.start(ElasticSearchSink.java:357)  
at org.apache.flume.sink.DefaultSinkProcessor.start(DefaultSinkProcessor.java:46)  
at org.apache.flume.SinkRunner.start(SinkRunner.java:79)  
at org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:251)  
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)  
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)  
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)  
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)  
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)  
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)  
at java.lang.Thread.run(Thread.java:745)

原因：

Elasticsearch的版本过高，导致Flume的jar包与Elasticsearch不兼容

解决方案：重置Elasticsearch版本至1.7.1

也可参照如下两篇文章的解决方案：

http://stackoverflow.com/questions/33732193/configure-sink-elasticsearch-apache-flume

https://github.com/elastic/elasticsearch/issues/14187

[Java]: http://lib.csdn.net/base/javase
[.NET]: http://lib.csdn.net/base/dotnet