import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction; //导入依赖的package包/类
public static void main(String[] args) throws Exception {

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<String> source = env.generateSequence(0, 20).map(new MapFunction<Long, String>() {
			@Override
			public String map(Long value) throws Exception {
				return "message #" + value;
			}
		});

		Map<String, String> userConfig = new HashMap<>();
		userConfig.put("cluster.name", "elasticsearch");
		// This instructs the sink to emit after every element, otherwise they would be buffered
		userConfig.put(ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "1");

		List<TransportAddress> transports = new ArrayList<>();
		transports.add(new InetSocketTransportAddress(InetAddress.getByName("127.0.0.1"), 9300));

		source.addSink(new ElasticsearchSink<>(userConfig, transports, new ElasticsearchSinkFunction<String>() {
			@Override
			public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
				indexer.add(createIndexRequest(element));
			}
		}));

		env.execute("Elasticsearch Sink Example");
	}
 

import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction; //导入依赖的package包/类
@Override
protected <T> ElasticsearchSinkBase<T> createElasticsearchSinkForEmbeddedNode(
	Map<String, String> userConfig, ElasticsearchSinkFunction<T> elasticsearchSinkFunction) throws Exception {

	List<InetSocketAddress> transports = new ArrayList<>();
	transports.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300));

	return new ElasticsearchSink<>(userConfig, transports, elasticsearchSinkFunction);
}
 

import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction; //导入依赖的package包/类
/**
 * Creates a new {@code ElasticsearchSink} that connects to the cluster using a {@link TransportClient}.
 *
 * @param userConfig The map of user settings that are used when constructing the {@link TransportClient} and {@link BulkProcessor}
 * @param transportAddresses The addresses of Elasticsearch nodes to which to connect using a {@link TransportClient}
 * @param elasticsearchSinkFunction This is used to generate multiple {@link ActionRequest} from the incoming element
 */
public ElasticsearchSink(
	Map<String, String> userConfig,
	List<InetSocketAddress> transportAddresses,
	ElasticsearchSinkFunction<T> elasticsearchSinkFunction) {

	this(userConfig, transportAddresses, elasticsearchSinkFunction, new NoOpFailureHandler());
}
 

import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction; //导入依赖的package包/类
public static void main(String[] args) throws Exception {

		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<String> source = env.generateSequence(0, 20).map(new MapFunction<Long, String>() {
			@Override
			public String map(Long value) throws Exception {
				return "message #" + value;
			}
		});

		Map<String, String> userConfig = new HashMap<>();
		userConfig.put("cluster.name", "elasticsearch");
		// This instructs the sink to emit after every element, otherwise they would be buffered
		userConfig.put(ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "1");

		List<InetSocketAddress> transports = new ArrayList<>();
		transports.add(new InetSocketAddress(InetAddress.getByName("127.0.0.1"), 9300));

		source.addSink(new ElasticsearchSink<>(userConfig, transports, new ElasticsearchSinkFunction<String>() {
			@Override
			public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
				indexer.add(createIndexRequest(element));
			}
		}));

		env.execute("Elasticsearch Sink Example");
	}
 

import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction; //导入依赖的package包/类
public ESSink(ESProperties config, ElasticsearchSinkFunction<T> function) {
  super(config, config.getTransportAddresses(), function);
}
 

import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction; //导入依赖的package包/类
@Override
protected <T> ElasticsearchSinkBase<T> createElasticsearchSink(Map<String, String> userConfig,
															List<InetSocketAddress> transportAddresses,
															ElasticsearchSinkFunction<T> elasticsearchSinkFunction) {
	return new ElasticsearchSink<>(userConfig, transportAddresses, elasticsearchSinkFunction);
}