ES写数据的基本程序

  1 package com.znv.elasticsearch;
  2 
  3 import org.elasticsearch.action.bulk.BulkRequestBuilder;
  4 import org.elasticsearch.action.bulk.BulkResponse;
  5 import org.elasticsearch.client.transport.TransportClient;
  6 import org.elasticsearch.common.settings.Settings;
  7 import org.elasticsearch.common.transport.InetSocketTransportAddress;
  8 import org.elasticsearch.transport.client.PreBuiltTransportClient;
  9 
 10 import java.net.InetAddress;
 11 import java.net.UnknownHostException;
 12 import java.text.SimpleDateFormat;
 13 import java.util.Date;
 14 import java.util.Random;
 15 
 16 import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
 17 
 18 
 19 public class esTestData {
 20     private static TransportClient client = null;
 21     private static Random rnd = new Random();
 22     private static SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 23     private static final String startDate = "2017-01-01 00:00:00";
 24 
 25 
 26     private static long randomNum(long begin, long end)
 27     {
 28         long rtn = begin + (long)(Math.random() * (end - begin));
 29         if (rtn == begin || rtn == end)
 30         {
 31             return randomNum(begin,end);
 32         }
 33         return rtn;
 34     } 
 35    public  static Date randomDate(String beginDate, String endDate)
 36     {
 37         try
 38         {
 39             SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 40             Date start = format.parse(beginDate);
 41             Date end = format.parse(endDate);
 42             if (start.getTime() >= end.getTime())
 43             {
 44                 return null;
 45             }
 46             long date = randomNum(start.getTime(),end.getTime());
 47             return new Date(date);
 48         }
 49         catch (Exception e)
 50         {
 51             e.printStackTrace();
 52         }
 53         return null;
 54     }
 55  
 56   public static void initClient() {
 57         try {
 58             // on startup
 59             Settings settings = Settings.builder()
 60                     .put("cluster.name", "lv08-elasticsearch").build();
 61             client = new PreBuiltTransportClient(settings)
 62                     .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("10.45.157.130"), 9300));
 63         } catch (UnknownHostException e) {
 64             e.printStackTrace();
 65         }
 66     }
 67 
 68     public static void closeClient() {
 69         client.close();
 70     }
 71 
 72  public static void bulkPutEs() {
 73         try {
 74 
 75             long t1 = System.currentTimeMillis();
 76             int n = 0;
 77             //写入10*1000条数据
 78             while (n++ < 10) {
 79                 BulkRequestBuilder bulkRequest = client.prepareBulk();
 80                 for (int i = 0; i < 1000; i++) {
 81                     int id = rnd.nextInt(1000);
 82                     String name = "名字" + id;
 83                     int gender = rnd.nextInt(2);
 84                     int age = rnd.nextInt(100);
 85                     Date enter_time = randomDate(startDate,df.format(new Date()));
 86                     bulkRequest.add(client.prepareIndex("test-total", "type1", String.valueOf(id)+String.valueOf(enter_time.getTime()))
 87                             .setSource(jsonBuilder()
 88                                             .startObject()
 89                                             .field("id", id)
 90                                             .field("name",name)
 91                                             .field("gender",gender)
 92                                             .field("age", age)
 93                                             .field("enter_time", enter_time)
 94                                             .endObject()
 95                             )
 96                     );
 97                 }
 98                BulkResponse bulkResponse = bulkRequest.get();
 99                 if (bulkResponse.hasFailures()) {
100                     // process failures by iterating through each bulk response item
101                     System.out.println(bulkRequest.toString());
102                 }
103             }
104 
105             long ts = System.currentTimeMillis() - t1;
106             System.out.println("cost "+ts +" ms.");
107 
108         } catch (Exception e) {
109             e.printStackTrace();
110         }
111     }
112 

113  public static void main(String []args) {
114         initClient();
115         bulkPutEs();
116         closeClient();
117     }
118 
119 }