实现了将Kafka中的数据推送给Mongodb，然后再将Mongodb中的数据推送给Es的过程。数据来源是来自txt文档中的180万条数据。

准备工作：

1）在Mongdb集群上创建一个数据库mydb，并创建一个空的Collection，命名为netflows

[java]  view plain  copy
  1. @SuppressWarnings("deprecation")  
  2.            Mongo mongo = new Mongo("10.10.16.251", 10111);    
  3.            
  4.            DB db = mongo.getDB("mydb");       
  5.              
  6.            //创建Collection，但是不添加数据  
  7.            db.createCollection("netflows", null);                                         
  8.            DBCollection dbColl = db.getCollection("netflows");  

2）在kafka的集群上创建一个主题flume1

[java]  view plain  copy
  1. bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic flume1  

2）在es集群上创建一个索引myindex，该索引的类型是netflows

[java]  view plain  copy
  1. IndexResponse res = client.prepareIndex().setIndex("myindex").setType("netflows").execute().actionGet();  

下面是代码实现：

1.从文件中读取测试数据，并推动给Kafka

[java]  view plain  copy
  1. package com.test;  
  2.   
  3. import java.io.BufferedReader;  
  4. import java.io.FileNotFoundException;  
  5. import java.io.FileReader;  
  6. import java.io.IOException;  
  7. import java.text.DateFormat;  
  8. import java.text.ParseException;  
  9. import java.text.SimpleDateFormat;  
  10. import java.util.Calendar;  
  11. import java.util.Date;  
  12. import java.util.Properties;  
  13.   
  14. import kafka.javaapi.producer.Producer;  
  15. import kafka.producer.KeyedMessage;  
  16. import kafka.producer.ProducerConfig;  
  17.   
  18. public class Kafka_Producer {  
  19.   
  20.     public static void main(String[] args) {  
  21.   
  22.         /* 
  23.          * Properties props = new Properties(); 
  24.          * props.setProperty("metadata.broker.list","10.10.16.253:9092"); 
  25.          * props.setProperty 
  26.          * ("serializer.class","kafka.serializer.StringEncoder"); 
  27.          * props.put("request.required.acks","-1"); 
  28.          *  
  29.          * ProducerConfig config = new ProducerConfig(props); Producer<String, 
  30.          * String> producer = new Producer<String, String>(config); 
  31.          * KeyedMessage<String, String> data = new KeyedMessage<String, 
  32.          * String>("flume","test-kafka"); 
  33.          *  
  34.          * producer.send(data); 
  35.          *  
  36.          * producer.close(); 
  37.          */  
  38.   
  39.         MessageSender messageSender = new MessageSender();  
  40.   
  41.         FileReader fr;  
  42.         try {  
  43.               
  44.             fr = new FileReader("C:\\TxtData\\NetFlowAttackDDOS\\4test.txt");  
  45.             BufferedReader br = new BufferedReader(fr);  
  46.             String line = "";  
  47.             String[] arrs = null;  
  48.               
  49.             DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");  
  50.             Date date = null;  
  51.             long num = 0;  
  52.               
  53.             while ((line = br.readLine()) != null) {  
  54.                   
  55.                 arrs = line.split(",");                          
  56.                 date = format.parse(format.format(new Date()));                   
  57.                 Calendar cla = Calendar.getInstance();  
  58.                 cla.setTime(date);  
  59.                 cla.add(Calendar.HOUR_OF_DAY,8);      
  60.                   
  61.                 messageSender.sendToKafkaTopic((num+10001) + ","+arrs[3] + " ," + arrs[4] + " ," + arrs[5]+ ","+arrs[6]);         
  62.                 num ++;                   
  63.             }  
  64.               
  65.             br.close();  
  66.             fr.close();  
  67.           
  68.         } catch (FileNotFoundException e) {  
  69.           
  70.             e.printStackTrace();  
  71.         } catch (IOException e) {  
  72.           
  73.             e.printStackTrace();  
  74.         } catch (ParseException e) {  
  75.             // TODO Auto-generated catch block  
  76.             e.printStackTrace();  
  77.         }  
  78.           
  79.         messageSender.close();  
  80.     }  
  81.   
  82. }  
其中MessageSender中代码如下所示：
[java]  view plain  copy
  1. package com.test;  
  2.   
  3. import java.util.Properties;  
  4.   
  5. import com.test.utils.ConstanUtil;  
  6. import com.test.utils.ObjectFormatUtil;  
  7.   
  8. import kafka.javaapi.producer.Producer;  
  9. import kafka.producer.KeyedMessage;  
  10. import kafka.producer.ProducerConfig;  
  11.   
  12. public class MessageSender {  
  13.   
  14.     private long windowStart = 0;  
  15.       
  16.     private long windowCurrent  = 0;  
  17.   
  18.       
  19.     private long sendSize = 0;  
  20.       
  21.     Producer<String, String> producer = null;   
  22.       
  23.     private StringBuilder records = new StringBuilder();  
  24.       
  25.     public MessageSender() {  
  26.       
  27.         initKafkaProperties();  
  28.     }  
  29.       
  30.     private void initKafkaProperties(){  
  31.           
  32.          Properties props = new Properties();     
  33.          props.setProperty("metadata.broker.list","10.10.16.253:9092,10.10.16.252:9092,10.10.16.249:9092");     
  34.           
  35.          props.setProperty("serializer.class","kafka.serializer.StringEncoder");     
  36.          props.put("request.required.acks","1");     
  37.          props.put("zookeeper.session.timeout.ms", "400000");  
  38.          ProducerConfig config = new ProducerConfig(props);     
  39.          producer = new Producer<String, String>(config);    
  40.     }  
  41.       
  42.     public void sendToKafkaTopic(String message){  
  43.           
  44.         KeyedMessage<String, String> data = new KeyedMessage<String, String>("flume1",message);   
  45.         producer.send(data);  
  46.           
  47.         /*long date = ObjectFormatUtil.formatDateToMinute(System.currentTimeMillis()); 
  48.         windowCurrent = date; 
  49.          
  50.         if(windowStart == 0){//初始化开始的时间窗口 
  51.              
  52.             windowStart = date; 
  53.         } 
  54.          
  55.         if ((windowCurrent - windowStart) >= (ConstanUtil.TIME_WINDOW_LEN)) { 
  56.              
  57.             if(records.length() != 0){ 
  58.                 String sendContent = records.substring(0,records.lastIndexOf("@")); 
  59.                  
  60.                 KeyedMessage<String, String> data = new KeyedMessage<String, String>("flume1",sendContent);    
  61.                 producer.send(data);                 
  62.                 sendContent = null; 
  63.                 records = new StringBuilder(); 
  64.                 sendSize = 0; 
  65.             } 
  66.         } 
  67.          
  68.         records.append(message + "@"); 
  69.         sendSize++; 
  70.          
  71.         if ((sendSize >= 100)) { 
  72.              
  73.             String sendContent = records.substring(0,records.lastIndexOf("@"));//去掉最后面的@符号 
  74.             KeyedMessage<String, String> data = new KeyedMessage<String, String>("flume1",sendContent);  
  75.             producer.send(data);             
  76.             sendContent = null; 
  77.             records = new StringBuilder(); 
  78.             sendSize = 0;//1000条记录向Kafka发送一次，发送完后将记录数据包的个数清零. 
  79.         }*/  
  80.     }  
  81.       
  82.     public void close(){  
  83.       
  84.         try {  
  85.               
  86.             if(null != producer){  
  87.                   
  88.                 producer.close();  
  89.             }  
  90.               
  91.         } catch (Exception e) {  
  92.               
  93.             e.printStackTrace();  
  94.         }  
  95.     }  
  96. }  
2.Kafka接收数据，并向Mongodb中存入，并同时同步给es的代码
[java]  view plain  copy
  1. package com.test.thread;  
  2.   
  3.   
  4. import java.net.UnknownHostException;  
  5. import java.text.DateFormat;  
  6. import java.text.ParseException;  
  7. import java.text.SimpleDateFormat;  
  8. import java.util.Calendar;  
  9. import java.util.Date;  
  10. import java.util.HashMap;  
  11. import java.util.List;  
  12. import java.util.Map;  
  13. import java.util.Properties;  
  14.   
  15.   
  16. import org.elasticsearch.action.get.GetResponse;  
  17. import org.elasticsearch.action.index.IndexResponse;  
  18. import org.elasticsearch.client.Client;  
  19. import org.elasticsearch.client.transport.TransportClient;  
  20. import org.elasticsearch.common.transport.InetSocketTransportAddress;  
  21. import com.mongodb.BasicDBObject;  
  22. import com.mongodb.DB;  
  23. import com.mongodb.DBCollection;  
  24. import com.mongodb.DBObject;  
  25. import com.mongodb.Mongo;  
  26. import com.test.model.NetFlow;  
  27. import com.test.utils.ESUtils;  
  28.   
  29.   
  30. import kafka.consumer.Consumer;  
  31. import kafka.consumer.ConsumerConfig;  
  32. import kafka.consumer.ConsumerIterator;  
  33. import kafka.consumer.KafkaStream;  
  34. import kafka.javaapi.consumer.ConsumerConnector;  
  35.   
  36.   
  37. public class ConsumerThread extends Thread {  
  38.   
  39.   
  40. <span style="white-space:pre">    </span>private String topic;  
  41. <span style="white-space:pre">    </span>private String mongodbIp;  
  42. <span style="white-space:pre">    </span>private int mongodbPort;  
  43. <span style="white-space:pre">    </span>private String indexName;  
  44. <span style="white-space:pre">    </span>private String indexType;  
  45. <span style="white-space:pre">    </span>private String dbName;  
  46. <span style="white-space:pre">    </span>private String collName;  
  47.   
  48.   
  49. <span style="white-space:pre">    </span>public ConsumerThread(String topic,String mongodbIp,int mongodbPort,String indexName,String indexType,String dbName,String collName) {  
  50. <span style="white-space:pre">        </span>  
  51. <span style="white-space:pre">        </span>this.topic = topic;  
  52. <span style="white-space:pre">        </span>this.mongodbIp = mongodbIp;  
  53. <span style="white-space:pre">        </span>this.mongodbPort = mongodbPort;  
  54. <span style="white-space:pre">        </span>this.indexName = indexName;  
  55. <span style="white-space:pre">        </span>this.indexType = indexType;  
  56. <span style="white-space:pre">        </span>this.dbName = dbName;  
  57. <span style="white-space:pre">        </span>this.collName = collName;  
  58. <span style="white-space:pre">        </span>  
  59. <span style="white-space:pre">    </span>}  
  60.   
  61.   
  62. <span style="white-space:pre">    </span>@SuppressWarnings({ "deprecation", "resource" })  
  63. <span style="white-space:pre">    </span>@Override  
  64. <span style="white-space:pre">    </span>public void run() {  
  65. <span style="white-space:pre">        </span>  
  66. <span style="white-space:pre">        </span>ConsumerConnector consumer = createConsumer();  
  67. <span style="white-space:pre">        </span>Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
  68. <span style="white-space:pre">        </span>topicCountMap.put(this.topic, 1); // 一次从主题中获取一个数据  
  69. <span style="white-space:pre">        </span>Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);  
  70. <span style="white-space:pre">        </span>KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据  
  71. <span style="white-space:pre">        </span>ConsumerIterator<byte[], byte[]> iterator = stream.iterator();  
  72. <span style="white-space:pre">        </span>  
  73. <span style="white-space:pre">        </span>Mongo mongo;  
  74. <span style="white-space:pre">        </span>@SuppressWarnings("unused")  
  75. <span style="white-space:pre">        </span>Client client;  
  76. <span style="white-space:pre">        </span>  
  77. <span style="white-space:pre">        </span>try {  
  78. <span style="white-space:pre">            </span>  
  79. <span style="white-space:pre">            </span>mongo = new Mongo(this.mongodbIp, this.mongodbPort);  
  80. <span style="white-space:pre">            </span>DB db = mongo.getDB(this.dbName);  
  81. <span style="white-space:pre">            </span>DBCollection dbColl = db.getCollection(this.collName);  
  82. <span style="white-space:pre">        </span>  
  83. <span style="white-space:pre">            </span>client = new TransportClient().addTransportAddress(new InetSocketTransportAddress("10.10.16.253", 9300));<span style="white-space:pre">              </span>  
  84. <span style="white-space:pre">            </span>  
  85. <span style="white-space:pre">            </span>while (iterator.hasNext()) {  
  86. <span style="white-space:pre">                </span>  
  87. <span style="white-space:pre">                </span>String message = new String(iterator.next().message());  
  88. <span style="white-space:pre">                </span>  
  89. <span style="white-space:pre">                </span>String[] recMessage = message.split(",");  
  90. <span style="white-space:pre">                </span>  
  91. <span style="white-space:pre">                </span>if(recMessage.length !=9){  
  92. <span style="white-space:pre">                    </span>  
  93. <span style="white-space:pre">                    </span>continue;  
  94. <span style="white-space:pre">                </span>}  
  95. <span style="white-space:pre">                </span>  
  96. <span style="white-space:pre">                </span>else{  
  97. <span style="white-space:pre">                    </span>  
  98. <span style="white-space:pre">                    </span>for (int j = 0;j < recMessage.length; j++) {  
  99.   
  100.   
  101. <span style="white-space:pre">                    </span>   DBObject data4 = new BasicDBObject();  
  102. <span style="white-space:pre">                    </span>   data4.put("_id", 1);  
  103. <span style="white-space:pre">                    </span>   data4.put("sourceIp", recMessage[3]);  
  104. <span style="white-space:pre">                    </span>   data4.put("sourcePort", recMessage[4]);  
  105. <span style="white-space:pre">                    </span>   data4.put("destIp", recMessage[5]);  
  106. <span style="white-space:pre">                    </span>   data4.put("destPort", recMessage[6]);  
  107. <span style="white-space:pre">    </span>                       
  108. <span style="white-space:pre">                    </span>   DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");  
  109. <span style="white-space:pre">                    </span>   Date date = null;  
  110. <span style="white-space:pre">                    </span>   date = format.parse(format.format(new Date()));  
  111.   
  112.   
  113. <span style="white-space:pre">                    </span>   Calendar cla = Calendar.getInstance();  
  114. <span style="white-space:pre">                    </span>   cla.setTime(date);  
  115. <span style="white-space:pre">                    </span>   cla.add(Calendar.HOUR_OF_DAY, 8);  
  116. <span style="white-space:pre">                    </span>   data4.put("update_time", cla.getTime().getTime());<span style="white-space:pre">                                 </span> <span style="white-space:pre">  </span>  
  117. <span style="white-space:pre">                    </span>   dbColl.insert(data4);  
  118. <span style="white-space:pre">                    </span>  <span style="white-space:pre">        </span>  
  119. <span style="white-space:pre">                    </span>   GetResponse getResponse = client.prepareGet().setIndex(this.indexName).setType(this.indexType).setId(recMessage[0]).execute().actionGet();  
  120. <span style="white-space:pre">                    </span>      
  121. <span style="white-space:pre">                    </span>   String searchResult = getResponse.getSourceAsString();  
  122. <span style="white-space:pre">                    </span>    
  123. <span style="white-space:pre">                    </span>   if(searchResult==null){  
  124. <span style="white-space:pre">                            </span>     
  125. <span style="white-space:pre">                        </span>    //向es中存入  
  126. <span style="white-space:pre">                    </span>    <span style="white-space:pre"> </span>NetFlow netFlow = new NetFlow();  
  127. <span style="white-space:pre">                            </span>netFlow.setSourceIp(recMessage[1]);  
  128. <span style="white-space:pre">                            </span>netFlow.setSourcePort(Integer.parseInt(recMessage[2].trim()));<span style="white-space:pre"> </span>  
  129. <span style="white-space:pre">                            </span>netFlow.setDestIp(recMessage[3]);<span style="white-space:pre">          </span>  
  130. <span style="white-space:pre">                            </span>netFlow.setDestPort(Integer.parseInt(recMessage[4].trim()));<span style="white-space:pre">                   </span>  
  131. <span style="white-space:pre">                            </span>String jsondata = ESUtils.toJson(netFlow);  
  132. <span style="white-space:pre">                    </span>    <span style="white-space:pre"> </span>  
  133. <span style="white-space:pre">                            </span>IndexResponse indexResponse = client.prepareIndex().setIndex(this.indexName).setType(this.indexType).setId(recMessage[0]).setSource(jsondata).execute().actionGet();  
  134. <span style="white-space:pre">                    </span>          
  135. <span style="white-space:pre">                    </span>   }  
  136. <span style="white-space:pre">                    </span>   if(searchResult!=null){  
  137. <span style="white-space:pre">                            </span>  
  138. <span style="white-space:pre">                        </span>   //什么操作都不做  
  139. <span style="white-space:pre">    </span>                   }  
  140. <span style="white-space:pre">                    </span>}  
  141. <span style="white-space:pre">                </span>}<span style="white-space:pre">      </span>  
  142. <span style="white-space:pre">            </span>}  
  143.   
  144.   
  145. <span style="white-space:pre">        </span>} catch (UnknownHostException e) {  
  146. <span style="white-space:pre">            </span>  
  147. <span style="white-space:pre">            </span>e.printStackTrace();<span style="white-space:pre">       </span>  
  148. <span style="white-space:pre">            </span>consumer.shutdown();  
  149. <span style="white-space:pre">            </span>  
  150. <span style="white-space:pre">        </span>} catch (ParseException e) {  
  151.   
  152.   
  153. <span style="white-space:pre">            </span>e.printStackTrace();  
  154. <span style="white-space:pre">            </span>consumer.shutdown();  
  155. <span style="white-space:pre">        </span>}  
  156. <span style="white-space:pre">        </span>  
  157. <span style="white-space:pre">        </span>  
  158. <span style="white-space:pre">    </span>}  
  159.   
  160.   
  161. <span style="white-space:pre">    </span>private ConsumerConnector createConsumer() {  
  162. <span style="white-space:pre">    </span>  
  163. <span style="white-space:pre">        </span>Properties properties = new Properties();  
  164. <span style="white-space:pre">        </span>properties.put("zookeeper.connect","10.10.16.252:2181,10.10.16.253:2181,10.10.16.249:2181");// 声明zk  
  165. <span style="white-space:pre">        </span>properties.put("group.id", "test-consummer-group");  
  166. <span style="white-space:pre">        </span>properties.put("serializer.class", "kafka.serializer.StringEncoder");  
  167. <span style="white-space:pre">        </span>return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));  
  168. <span style="white-space:pre">    </span>}  
  169.   
  170.   
  171. <span style="white-space:pre">    </span>public static void main(String[] args) {  
  172.   
  173.   
  174. <span style="white-space:pre">        </span>new ConsumerThread("flume1","10.10.16.251",10111,"myindex","netflows","mydb","netflows").start();// 使用kafka集群中创建好的主题 test  
  175.   
  176.   
  177. <span style="white-space:pre">    </span>}  
  178.   
  179.   
  180. }  
其中NetFlow是个实体类，如下：
[java]  view plain  copy
  1. package com.test.model;  
  2.   
  3. public class NetFlow {  
  4.   
  5.       
  6.     private String sourceIp;  
  7.     private int sourcePort;  
  8.     private String destIp;  
  9.     private int destPort;  
  10.     public String getSourceIp() {  
  11.         return sourceIp;  
  12.     }  
  13.     public void setSourceIp(String sourceIp) {  
  14.         this.sourceIp = sourceIp;  
  15.     }  
  16.     public int getSourcePort() {  
  17.         return sourcePort;  
  18.     }  
  19.     public void setSourcePort(int sourcePort) {  
  20.         this.sourcePort = sourcePort;  
  21.     }  
  22.     public String getDestIp() {  
  23.         return destIp;  
  24.     }  
  25.     public void setDestIp(String destIp) {  
  26.         this.destIp = destIp;  
  27.     }  
  28.     public int getDestPort() {  
  29.         return destPort;  
  30.     }  
  31.     public void setDestPort(int destPort) {  
  32.         this.destPort = destPort;  
  33.     }  
  34.       
  35. }  
ESUtils是一个工具类：

[java]  view plain  copy
  1. package  com.test.utils;  
  2.   
  3. import java.io.IOException;  
  4.   
  5. import org.codehaus.jackson.JsonProcessingException;  
  6. import org.codehaus.jackson.map.ObjectMapper;  
  7.   
  8. /** 
  9.  * 使用jackson定义了一个将对象转化成json的工具类 
  10.  */  
  11. public class ESUtils {    
  12.     private static ObjectMapper objectMapper = new ObjectMapper();    
  13.     public static String toJson(Object o){    
  14.         try {    
  15.             return objectMapper.writeValueAsString(o);  
  16.         } catch (JsonProcessingException e) {    
  17.             e.printStackTrace();    
  18.         } catch (IOException e) {  
  19.             e.printStackTrace();  
  20.         }    
  21.         return "";    
  22.     }    
  23. }  
Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南