flume1.6自定义elasticsearch2.3的sink插件

我会带着你远行 2021-09-14 10:52 168阅读 0赞

目前官方Flume最新的版本仍然不支持Elasticsearch2.x版本，flume的老版本（1.6）只支持Elasticsearch1.7.x的版本，因为Elasticsearch2.x版本做了比较大的改动，很多API都已经废弃不用了。

在github上找了一个项目：https://github.com/lucidfrontier45/ElasticsearchSink2  也是按照flume源码中sink的模式编写的，其实也很简单。最近，公司有个项目也是需要将日志手机到es上，供实时查询，索性自己写了一个。

一、整体架构：

1、线上业务代码将日志数据发送到kafka，数据结构是pb，kafka中存放的是byte流；

2、flume-source对接kafka，自定义一个intercept，将byte流解成结构化数据；

3、自定义es-sink，将结构化数据存放到es中。

二、代码：

1、pb：

option java_package = "XXX.base.proto";
    option java_outer_classname="ApiLogPB";
    
    message ApiLog {
        optional string puid = 1;
        optional string uId = 2;
        optional string reqId = 3;
        optional int32 fNum = 4;
        optional int32 cost = 5;
        optional string chId = 6 [ default = "default"];
        optional string strategy = 7;
        repeated int64 recId = 8;
        optional string txt = 9;
        optional string vedio = 10;
        optional string gallery = 11;
        optional string pMap = 12;
        optional string paramMap = 13;
    }

2、线上发送数据到kafka：

public Integer call() throws Exception {
    		Builder newBuilder = ApiLogPB.ApiLog.newBuilder();
    		
    		newBuilder.setReqId(reqId);
    		newBuilder.setUId(uId);
    		newBuilder.setPuid(puid);
    		newBuilder.setChId(channelId);
    		newBuilder.setFNum(fNum);
    		newBuilder.setCost(Integer.parseInt((System.currentTimeMillis()-a)+""));
    		newBuilder.setStrategy(personalResponse.getStrategy());
    		
    		List<Long> recIdSet = personalResponse.getRecIdSet();
    		for (Long fid:recIdSet) {
    			newBuilder.addRecId(fid);
    		}
    		
    		Map<String, List<Long>> typeFeeds = personalResponse.getTypeFeeds();
    		List<Long> newList = typeFeeds.get("NEWS");
    		newBuilder.setTxt(newList==null ? "":newList.toString());
    		List<Long> garleryList = typeFeeds.get("GALLERY");
    		newBuilder.setGallery(garleryList==null ? "":garleryList.toString());
    		List<Long> vedioList = typeFeeds.get("VIDEO");
    		newBuilder.setVedio(vedioList==null?"":vedioList.toString());
    		
    		Map<String, List<Long>> mutilFeeds = personalResponse.getMutilFeeds();
    		StringBuilder pSb = new StringBuilder();
    		Set<Entry<String, List<Long>>> entrySet = mutilFeeds.entrySet();
    		int i = 0;
    		for (Entry<String, List<Long>> engry : entrySet) {
    			String key = engry.getKey();
    			List<Long> value = engry.getValue();
    			if (CollectionUtils.isNotEmpty(value)) {
    				String v1 = StringUtils.strip(value.toString(),"[]");
    				
    				pSb.append(key).append(":").append(v1);
    				if (i<entrySet.size()-1) {
    					pSb.append(";");
    				}
    				i++;
    			}
    		}
    		newBuilder.setPMap(pSb.toString());
    		//newBuilder.setPMap(mutilFeeds.toString());
    		
    		newBuilder.setParamMap(paraMap.toString());
    		
    		KafkaDao.getInstance().sendMessage(uId,newBuilder.build().toByteArray());
    		return null;
    	}

3、日志实体类：

package XXX.log.common.entity;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import com.google.protobuf.InvalidProtocolBufferException;
    import com.iqiyi.ttbrain.base.proto.ApiLogPB;
    import com.iqiyi.ttbrain.base.proto.ApiLogPB.ApiLog;
    
    
    public class ApiEntity {
    	public static final int P_N_SIZE = 50;
    	private static final String flag = "\t";
    	
    	
    	private String reqId="";
    	private String uid="";
    	private String ppuid="";
    	private String channel="";
    	private int feedNum=10;
    	private int cost=0;
    	private String strategy="";
    	
    	private String timeStamp="";
    	private String host="";
    	
    	private String recFeedId="";
    	private String txt="";
    	private String gallery="";
    	private String vedio="";
    	
    	private String pMap = "";
    	private String paraMap = "";
    	
    	
    	@Override
    	public String toString() {
    		StringBuilder sb = new StringBuilder(768);
    		sb.append(reqId).append(flag).append(uid).append(flag).append(ppuid).append(flag).append(channel).append(flag)
    		.append(feedNum).append(flag).append(cost).append(flag).append(strategy).append(flag)
    		
    		.append(timeStamp).append(flag).append(host).append(flag)
    		
    		.append(recFeedId).append(flag)
    		.append(txt).append(flag).append(gallery).append(flag).append(vedio).append(flag)
    		
    		.append(pMap).append(flag)
    		.append(paraMap).append(flag)
    	 
    		.append("end");
    		
    		return sb.toString();
    	}
    	
    	public static ApiEntity parseFromPB(byte[] bytes) throws InvalidProtocolBufferException{
    		ApiEntity an = null;
    		if (bytes != null) {
    			an = new ApiEntity();
    			ApiLog apiLog = ApiLogPB.ApiLog.parseFrom(bytes);
    			
    			an.setReqId(apiLog.getReqId());
    			an.setUid(apiLog.getUId());
    			an.setPpuid(apiLog.getPuid());
    			an.setChannel(apiLog.getChId());
    			an.setFeedNum(apiLog.getFNum());
    			an.setCost(apiLog.getCost());
    			an.setStrategy(apiLog.getStrategy());
    			
    			StringBuilder sb = new StringBuilder();
    			List<Long> recIdList = apiLog.getRecIdList();
    			for (int i=0;i<recIdList.size();i++) {
    				sb.append(recIdList.get(i));
    				if (i<recIdList.size()-1) {
    					sb.append(",");
    				}
    			}
    			an.setRecFeedId(sb.toString());
    			
    			an.setTxt(apiLog.getTxt());
    			an.setGallery(apiLog.getGallery());
    			an.setVedio(apiLog.getVedio());
    			
    			an.setParaMap(apiLog.getParamMap());
    			an.setpMap(apiLog.getPMap());
    		}
    		
    		return an;
    	}
    	
    	public static Map<String,Object> parseToMap(String line) {
    		Map<String,Object> map = new HashMap<>();
    		if (line != null) {
    			String[] split = line.split(flag);
    			if (split.length>=15) {
    				map.put("reqId", split[0]);
    				map.put("uid", split[1]);
    				map.put("ppuid", split[2]);
    				map.put("channel", split[3]);
    				map.put("feedNum", split[4]);
    				map.put("cost", split[5]);
    				map.put("strategy", split[6]);
    				map.put("timeStamp", split[7]);
    				map.put("host", split[8]);
    				map.put("recFeedId", split[9]);
    				map.put("txt", split[10]);
    				map.put("gallery", split[11]);
    				map.put("vedio", split[12]);
    				map.put("pMap", split[13]);
    				map.put("paraMap", split[14]);
    			}
    		}
    		/*BeanMap beanMap = BeanMap.create(an);  
            for (Object key : beanMap.keySet()) {
            	map.put((String)key, beanMap.get(key));
            }*/
    
    		return map;
    	}
    	
    	
    	public static void main(String...strings) {
    		ApiEntity le = new ApiEntity();
    		le.setChannel("default");
    		le.setParaMap("sdfsdfdsfdsfds");
    		le.setpMap("12321321321321");
    		String line = le.toString();
    		System.out.println(line);
    		
    		String[] split = line.split(flag);
    		System.out.println(split.length);
    		
    		Map<String, Object> parseToMap = parseToMap(line);
    		System.out.println(parseToMap);
    		
    		System.out.println(split[15]);
    	}
    }

4、flume-intercept：

package XXX.log.flume.interceptor;
    
    import java.util.List;
    import java.util.Map;
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.interceptor.Interceptor;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import com.google.common.collect.Lists;
    import com.iqiyi.ttbrain.log.common.entity.ApiEntity;
    
    
    public class ApiInterceptor implements Interceptor {
    	private static final Logger logger = LoggerFactory.getLogger(ApiInterceptor.class);
    	
    	@Override
    	public void close() {
    		// TODO Auto-generated method stub
    		logger.info("flume ApiInterceptor is close");
    	}
    
    	@Override
    	public void initialize() {
    		// TODO Auto-generated method stub
    		logger.info("flume ApiInterceptor is initialize");
    	}
    
    	@Override
    	public Event intercept(Event event) {
    		try {
    			Map<String, String> headers = event.getHeaders();
    			byte[] body = event.getBody();
    			if (body != null) {
    				ApiEntity apiLog = null;
    				try{
    				apiLog = ApiEntity.parseFromPB(body);
    				}catch(Exception e){
    					logger.info("apiLog:{}",apiLog);
    				}
    				if (apiLog != null) {
    					String hostName = headers.get("hostname");
    					String timeStamp = headers.get("timestamp");
    					apiLog.setHost(hostName);
    					apiLog.setTimeStamp(timeStamp);
    //					logger.info(apiLog.toString());
    					event.setBody(apiLog.toString().getBytes());
    					return event;
    				}
    			}
    		} catch (Exception e ) {
    			logger.error("intercept:",e);
    		}
    		return null;
    	}
    	
    	@Override
    	public List<Event> intercept(List<Event> events) {
    		List<Event> intercepted = Lists.newArrayListWithCapacity(events.size());
    		for (Event event : events) {
                Event interceptedEvent = intercept(event);
                if (interceptedEvent != null) {
                    intercepted.add(interceptedEvent);
                }
            }
            return intercepted;
    	}
    	
    	public static class Builder implements Interceptor.Builder {
            //使用Builder初始化Interceptor
            @Override
            public Interceptor build() {
                return new ApiInterceptor();
            }
    
    		@Override
    		public void configure(Context arg0) {
    			// TODO Auto-generated method stub
    			
    		}
        }
    }

5、esDao工具类：

package XXX.log.flume.sink.db;
    
    import java.net.InetSocketAddress;
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import org.elasticsearch.action.bulk.BulkRequestBuilder;
    import org.elasticsearch.action.bulk.BulkResponse;
    import org.elasticsearch.action.index.IndexResponse;
    import org.elasticsearch.client.transport.TransportClient;
    import org.elasticsearch.common.settings.Settings;
    import org.elasticsearch.common.transport.InetSocketTransportAddress;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    
    public class EsDao {
    	private static final Logger logger = LoggerFactory.getLogger(EsDao.class);
    	// ip:port
    	private static final String clusterHost = "1.1.1.1:9300";
    	private static final String clusterName = "test";
    
    	private static TransportClient transportClient = null;
    
    	static {
    		Settings settings = Settings.settingsBuilder()
    				.put("cluster.name", clusterName).build();
    
    		transportClient = TransportClient.builder().settings(settings).build();
    
    		String[] hostNames = clusterHost.split(",");
    		InetSocketTransportAddress[] serverAddresses = new InetSocketTransportAddress[hostNames.length];
    		for (int i = 0; i < hostNames.length; i++) {
    			String[] hostPort = hostNames[i].trim().split(":");
    			String host = hostPort[0].trim();
    			int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1]
    					.trim()) : 9300;
    			serverAddresses[i] = new InetSocketTransportAddress(
    					new InetSocketAddress(host, port));
    		}
    
    		for (InetSocketTransportAddress host : serverAddresses) {
    			transportClient.addTransportAddress(host);
    		}
    
    	}
    
    	private EsDao() {
    	}
    	
    	public static void closeClient() {
    		transportClient.close();
    	}
    	
    	
    	/**
    	 * bulk——在一个请求中添加、更新和删除多个文档
    	 * @param indexName
    	 * @param indexType
    	 * @param datas
    	 * @return
    	 */
    	public static boolean bulk(String indexName,String indexType,List<Map<String,Object>> datas){
    		BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk();
    		for (Map<String,Object> map: datas) {
    			bulkRequestBuilder.add(transportClient.prepareIndex(indexName,
    					indexType).setSource(map));
    		}
    		BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
    		
    		return bulkResponse.hasFailures();
        }
    }

6、esSink：

package XXX.log.flume.sink;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import org.apache.flume.Channel;
    import org.apache.flume.Context;
    import org.apache.flume.Event;
    import org.apache.flume.EventDeliveryException;
    import org.apache.flume.Transaction;
    import org.apache.flume.conf.Configurable;
    import org.apache.flume.sink.AbstractSink;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import com.google.common.collect.Lists;
    import com.iqiyi.ttbrain.log.flume.sink.db.EsDao;
    
    
    public class EsSink extends AbstractSink implements Configurable {
    	
    	private static final Logger logger = LoggerFactory.getLogger(EsSink.class);
    	
    	private String indexName = "api";
    	private String indexType =  "api_type";
    	private int batchSize = 100;
    	
    	
    	public EsSink() {
    		logger.info("EsSink start...");
        }
    	
    	@Override
        public void start() {
            super.start();
            
        }
    	
    	@Override
        public void stop() {
            super.stop();
            
            //DataSourceUtils.closeDs();
        }
    	
    	@Override
    	public Status process() throws EventDeliveryException {
    		Status result = Status.READY;
            Transaction transaction = null;
            Event event = null;
            String content = "";
            List<Map<String,Object>> actions = Lists.newArrayList();
            
            try {
            	//flume
            	Channel channel = getChannel();
            	transaction = channel.getTransaction();
            	transaction.begin();
            	
            	for (int i = 0; i < batchSize; i++) {
            		event = channel.take();
                	if (event == null) {
                		result = Status.BACKOFF;
                		break;
                	} else {
                		content = new String(event.getBody(),"UTF-8");
                		String[] split = content.split("\t");
                		if (split.length < 15) {
                			continue;
                		}
                		//logger.info(content);
                		actions.add(generateMap(content));
                	}
            	}
            	
            	if (actions.size() > 0) {
                    //logger.info(actions.get(0));
            		EsDao.bulk(indexName, indexType, actions);
                }
            	
                transaction.commit();
            } catch (Throwable e) {
                /*try {
                	if (transaction != null) {
                		transaction.rollback();
                	}
                } catch (Exception e2) {
                	logger.error("flume transaction rollback error.", e2);
                }*/
                logger.error("Failed to commit flume transaction," +"Transaction rolled back.", e);
            } finally {
            	if (transaction != null) {
            		transaction.close();
            	}
            }
            return result;
    	}
    
    	@Override
    	public void configure(Context context) {
    		
    	}
    	
    	private static Map<String,Object> generateMap(String line) {
    		Map<String,Object> map = new HashMap<>();
    		if (line != null) {
    			String[] split = line.split("\t");
    			if (split.length>=15) {
    				map.put("reqId", split[0]);
    				map.put("uid", split[1]);
    				map.put("ppuid", split[2]);
    				map.put("channel", split[3]);
    				map.put("feedNum", split[4]);
    				map.put("cost", split[5]);
    				map.put("strategy", split[6]);
    				map.put("timeStamp", split[7]);
    				map.put("host", split[8]);
    				map.put("recFeedId", split[9]);
    				map.put("txt", split[10]);
    				map.put("gallery", split[11]);
    				map.put("vedio", split[12]);
    				map.put("pMap", split[13]);
    				map.put("paraMap", split[14]);
    			}
    		}
    		/*BeanMap beanMap = BeanMap.create(an);  
            for (Object key : beanMap.keySet()) {
            	map.put((String)key, beanMap.get(key));
            }*/
    
    		return map;
    	}
    }

三、flume配置：  
1）将上面的代码打包，放到flume的lib下

2）flume的配置如下：

agent1.sources = logsource
    agent1.channels = mc1 mc2
    agent1.sinks = avro-sink sink2
    
    
    agent1.sources.logsource.channels = mc1 mc2
    agent1.sinks.avro-sink.channel = mc1
    agent1.sinks.sink2.channel = mc2
    
    
    #source
    agent1.sources.logsource.type = org.apache.flume.source.kafka.KafkaSource
    agent1.sources.logsource.zookeeperConnect = ttAlgorithm-kafka-online001-jyltqbs.qiyi.virtual:2181,ttAlgorithm-kafka-online002-jyltqbs.qiyi.virtual:2181,ttAlgorithm-kafka-online003-jyltqbs.qiyi.virtual:2181,ttAlgorithm-kafka-online004-jyltqbs.qiyi.virtual:2181,ttAlgorithm-kafka-online005-jyltqbs.qiyi.virtual:2181
    agent1.sources.logsource.topic = topic_predict
    agent1.sources.logsource.groupId = flume
    agent1.sources.logsource.kafka.consumer.timeout.ms = 100
    
    #interceptor
    agent1.sources.logsource.interceptors=filt2 filt3 filt4
    agent1.sources.logsource.interceptors.filt2.type=host
    agent1.sources.logsource.interceptors.filt2.hostHeader=hostname
    agent1.sources.logsource.interceptors.filt2.useIP=true
    agent1.sources.logsource.interceptors.filt3.type=timestamp
    agent1.sources.logsource.interceptors.filt4.type=com.iqiyi.ttbrain.log.flume.interceptor.PredictInterceptor$Builder
    
    agent1.sources.logsource.selector.type = multiplexing
    agent1.sources.logsource.selector.header = isCool
    agent1.sources.logsource.selector.mapping.0 = mc1
    agent1.sources.logsource.selector.mapping.1 = mc2
    agent1.sources.logsource.selector.default = mc1
    
    #channel1
    agent1.channels.mc1.type = memory
    agent1.channels.mc1.capacity = 10000
    agent1.channels.mc1.transactionCapacity = 10000
    agent1.channels.mc1.keep-alive = 60
    
    #channel2
    agent1.channels.mc2.type = memory
    agent1.channels.mc2.capacity = 10000
    agent1.channels.mc2.transactionCapacity = 10000
    agent1.channels.mc2.keep-alive = 60
    
    
    #sink1
    #agent1.sinks.avro-sink.type = file_roll
    #agent1.sinks.avro-sink.sink.directory = /data/mysink
    #agent1.sinks.avro-sink.sink.rollInterval = 10000000
    agent1.sinks.avro-sink.type = hdfs
    agent1.sinks.avro-sink.hdfs.path = hdfs://hadoop-jy-namenode/data/qytt/flume/ttengine_predict/dt=%Y-%m-%d/hour=%H/
    agent1.sinks.avro-sink.hdfs.writeFormat = Text
    agent1.sinks.avro-sink.hdfs.fileType = DataStream
    agent1.sinks.avro-sink.hdfs.fileSuffix = .log
    agent1.sinks.avro-sink.hdfs.filePrefix = %Y-%m-%d_%H
    agent1.sinks.avro-sink.hdfs.rollInterval = 3600
    agent1.sinks.avro-sink.hdfs.rollSize = 0
    agent1.sinks.avro-sink.hdfs.rollCount = 0
    agent1.sinks.avro-sink.hdfs.batchSize = 1000
    agent1.sinks.avro-sink.hdfs.callTimeout = 60000
    agent1.sinks.avro-sink.hdfs.appendTimeout = 60000
    
    #sink2
    agent1.sinks.sink2.type = hdfs
    agent1.sinks.sink2.hdfs.path = hdfs://hadoop-jy-namenode/data/qytt/flume/ttengine_predict_cool_start/dt=%Y-%m-%d/hour=%H/
    agent1.sinks.sink2.hdfs.writeFormat = Text
    agent1.sinks.sink2.hdfs.fileType = DataStream
    agent1.sinks.sink2.hdfs.fileSuffix = .log
    agent1.sinks.sink2.hdfs.filePrefix = %Y-%m-%d_%H
    agent1.sinks.sink2.hdfs.rollInterval = 3600
    agent1.sinks.sink2.hdfs.rollSize = 0
    agent1.sinks.sink2.hdfs.rollCount = 0
    agent1.sinks.sink2.hdfs.batchSize = 1000
    agent1.sinks.sink2.hdfs.callTimeout = 60000
    agent1.sinks.sink2.hdfs.appendTimeout = 60000

3）注意事项：

flume1.6和es2.3版本有jar冲突问题，在编写flume-es-sink时，需要引入：

<dependency>  
            <groupId>org.elasticsearch</groupId>  
            <artifactId>elasticsearch</artifactId>  
            <version>2.3.2</version>  
    </dependency>

然后发现会有java.lang.NoSuchMethodError:com.google.common.util.concurrent.MoreExecutors.directExecutor()Ljava/util/concu‌rrent/Executor 报错，和 Exception in thread "main" java.lang.NoSuchFieldError: FAIL\_ON\_SYMBOL\_HASH\_OVERFLOW 两个问题，解决方法，在maven中加上

<dependency>  
        <groupId>com.google.guava</groupId>  
        <artifactId>guava</artifactId>  
        <version>18.0</version>  
    </dependency>

和

<dependency>    
         <groupId>com.fasterxml.jackson.core</groupId>    
         <artifactId>jackson-core</artifactId>    
         <version>2.6.2</version>    
     </dependency>

同时，在把jar包放到flume的lib中，启动flume也会报上面的错误，这时需要把flume的lib目录下guava-11.0.2.jar和jackson-core-2.3.1.jar 两个包去掉。  
  
具体可以参考：http://blog.csdn.net/liuxiao723846/article/details/78531916

参考文档：

http://tech.lede.com/2017/02/08/rd/server/flumeToEs/