import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
private void FlumeESSinkAndTestData(List<Event> searchEvents)
		throws EventDeliveryException, IOException, FileNotFoundException {
	flumeESSinkService.processEvents(searchEvents);

	Client client = searchClientService.getClient();
	client.admin().indices().refresh(Requests.refreshRequest()).actionGet();

	String indexName = "recentlyviewed" + '-'
			+ ElasticSearchIndexRequestBuilderFactory.df.format(new Date());
	long totalCount = client.prepareCount(indexName).get().getCount();
	System.out.println("Search total count is: " + totalCount);

	SearchHits hits = client.prepareSearch(indexName).get().getHits();
	System.out.println("Total hits: " + hits.getTotalHits());
	for (SearchHit searchHit : hits) {
		System.out.println(searchHit.getSource());
	}
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
@Test
public void testProcessEvents() {
	int searchEventsCount = 101;
	List<Event> searchEvents = generateSearchAnalyticsDataService
			.getSearchEvents(searchEventsCount);

	flumeESSinkService.processEvents(searchEvents);

	Client client = searchClientService.getClient();
	client.admin().indices().refresh(Requests.refreshRequest()).actionGet();

	String indexName = "recentlyviewed" + '-'
			+ ElasticSearchIndexRequestBuilderFactory.df.format(new Date());
	long totalCount = client.prepareCount(indexName).get().getCount();
	System.out.println("Search total count is: " + totalCount);

	SearchHits hits = client.prepareSearch(indexName).get().getHits();
	System.out.println("Total hits: " + hits.getTotalHits());
	for (SearchHit searchHit : hits) {
		System.out.println(searchHit.getSource());
	}

}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
/**
 *
 * @param clientType
 *    String representation of client type
 * @param hostNames
 *    Array of strings that represents hostnames with ports (hostname:port)
 * @param clusterName
 *    Elasticsearch cluster name used only by Transport Client
 * @param serializer
 *    Serializer of flume events to elasticsearch documents
 * @return
 */
public ElasticSearchClient getClient(String clientType, String[] hostNames,
    String clusterName, ElasticSearchEventSerializer serializer,
    ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
  if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
    return new ElasticSearchTransportClient(hostNames, clusterName, serializer);
  } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) { 
    return new ElasticSearchTransportClient(hostNames, clusterName, indexBuilder);
  } else if (clientType.equalsIgnoreCase(RestClient) && serializer != null) {
    return new ElasticSearchRestClient(hostNames, serializer);
  }
  throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
/**
 * Used for tests only. Creates local elasticsearch instance client.
 *
 * @param clientType Name of client to use
 * @param serializer Serializer for the event
 * @param indexBuilder Index builder factory
 *
 * @return Local elastic search instance client
 */
public ElasticSearchClient getLocalClient(String clientType,
                                          ElasticSearchEventSerializer serializer,
                                          ElasticSearchIndexRequestBuilderFactory indexBuilder)
    throws NoSuchClientTypeException {
  if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
    return new ElasticSearchTransportClient(serializer);
  } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null)  {
    return new ElasticSearchTransportClient(indexBuilder);
  } else if (clientType.equalsIgnoreCase(RestClient)) {
  }
  throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
/**
 * Used for testing
 */
public ElasticSearchTransportClient(Client client,
                                    ElasticSearchIndexRequestBuilderFactory requestBuilderFactory)
    throws IOException {
  this.client = client;
  requestBuilderFactory.createIndexRequest(client, null, null, null);
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
/**
 *
 * @param clientType String representation of client type
 * @param hostNames Array of strings that represents hosntames with ports
 * (hostname:port)
 * @param clusterName Elasticsearch cluster name used only by Transport
 * Client
 * @param serializer Serializer of flume events to elasticsearch documents
 * @return
 */
public ElasticSearchClient getClient(String clientType, String[] hostNames,
        String clusterName, ElasticSearchEventSerializer serializer,
        ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
    if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
        return new ElasticSearchTransportClient(hostNames, clusterName, serializer);
    } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
        return new ElasticSearchTransportClient(hostNames, clusterName, indexBuilder);
    } else if (clientType.equalsIgnoreCase(RestClient) && serializer != null) {
        return new ElasticSearchRestClient(hostNames, serializer);
    }
    throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
/**
 * Used for tests only. Creates local elasticsearch instance client.
 *
 * @param clientType Name of client to use
 * @param serializer Serializer for the event
 * @param indexBuilder Index builder factory
 *
 * @return Local elastic search instance client
 */
public ElasticSearchClient getLocalClient(String clientType, ElasticSearchEventSerializer serializer,
        ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
    if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
        return new ElasticSearchTransportClient(serializer);
    } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
        return new ElasticSearchTransportClient(indexBuilder);
    } else if (clientType.equalsIgnoreCase(RestClient)) {
    }
    throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
public ElasticSearchTransportClient(String[] hostNames, String clusterName,
        ElasticSearchIndexRequestBuilderFactory indexBuilder) {
    configureHostnames(hostNames);
    this.indexRequestBuilderFactory = indexBuilder;
    openClient(clusterName);
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
void initDefaults() {
	timestampedIndexName = DEFAULT_INDEX_NAME + '-'
			+ ElasticSearchIndexRequestBuilderFactory.df.format(FIXED_TIME_MILLIS);
	dateFormat = FastDateFormat.getInstance("yyyy.MM.dd", TimeZone.getTimeZone("Etc/UTC"));
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
private String getIndexName(String indexPrefix, long timestamp) {
	return new StringBuilder(indexPrefix).append('-')
			.append(ElasticSearchIndexRequestBuilderFactory.df
					.format(timestamp)).toString();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory; //导入依赖的package包/类
/**
 * Gets the name of the index to use for an index request
 * 
 * @return index name of the form 'indexPrefix-formattedTimestamp'
 * @param indexPrefix
 *            Prefix of index name to use -- as configured on the sink
 * @param timestamp
 *            timestamp (millis) to format / use
 */
private String getIndexName(String indexPrefix, long timestamp) {
	return new StringBuilder(indexPrefix)
			.append('-')
			.append(ElasticSearchIndexRequestBuilderFactory.df
					.format(timestamp)).toString();
}