BulkJSONDataESSink extends SimpleJSONDataESSink { protected String getBulkData(List<SimpleDataEvent> events) { StringBuilder builder = new StringBuilder(); for (SimpleDataEvent event : events) { builder.append(JSONUtils.getElasticSearchBulkHeader(event, this.indexName, this.typeName)); builder.append("\n"); builder.append(JSONUtils.getElasticSearchAddDocument(event.pairs())); builder.append("\n"); } return builder.toString(); }  @Override void init(PlayerConfig config); @Override boolean write(SimpleDataEvent event); static final String ES_BATCH_SIZE_KEY; static final String ES_BULK_END_POINT; }
@Test public void testGetBulkData() throws Exception { BulkJSONDataESSink sink = new BulkJSONDataESSink(); Map<String, String> properties = new HashMap<String, String>(); properties.put(SimpleJSONDataESSink.ES_BASE_URL_KEY, "http: properties.put(SimpleJSONDataESSink.ES_INDEX_NAME_KEY, "test"); properties.put(SimpleJSONDataESSink.ES_TYPE_NAME_KEY, "doc"); sink.init(new PlayerConfig(properties)); SimpleDataEvent event = new SimpleDataEvent("1"); event.addPair("name", "abc"); event.addPair("category", "it"); List<SimpleDataEvent> events = new ArrayList<SimpleDataEvent>(1); events.add(event); assertEquals( "{\"index\":{\"_index\":\"test\",\"_type\":\"doc\",\"_id\":\"1\"}}\n{\"category\":\"it\",\"name\":\"abc\"}\n", sink.getBulkData(events)); }