JarCastingflink-elasticsearch2-connector
Flink DataSet ElasticSearchOutputFormat create by https://www.iteblog.com based on org.apache.flink#flink-connector-elasticsearch2_2.10#1.1.2, We can use it in Scala or Java. Through the ElasticSearchOutputFormat, we can save Flink DataSet to elasticsearch.
Usage
Environment
Elasticsearch: 2.x.x
Flink: 1.x.x
Scala: 2.10.x
pom.xml
<dependency>
<groupId>com.iteblog</groupId>
<artifactId>flink-elasticsearch2-connector</artifactId>
<version>1.0.1</version>
</dependency>
Using in Scala
import scala.collection.JavaConversions._
val config = Map("bulk.flush.max.actions" -> "1000", "cluster.name" -> "elasticsearch")
val hosts = "www.iteblog.com"
val transports = hosts.split(",").map(host => new InetSocketAddress(InetAddress.getByName(host), 9300)).toList
val data : DataSet[String] = ....
data.output(new ElasticSearchOutputFormat(config, transports, new ElasticsearchSinkFunction[String] {
def createIndexRequest(element: String): IndexRequest = {
Requests.indexRequest.index("iteblog").`type`("info").source(element)
}
override def process(element: String, ctx: RuntimeContext, indexer: RequestIndexer) {
indexer.add(createIndexRequest(element))
}
}))
Using in Java
Map<String, String> config = new HashMap<>();
config.put("bulk.flush.max.actions", "1000");
config.put("cluster.name", "elasticsearch");
String hosts = "www.iteblog.com";
List<InetSocketAddress> list = Lists.newArrayList();
for (String host : hosts.split(",")) {
list.add(new InetSocketAddress(InetAddress.getByName(host), 9300));
}
DataSet<String> data = ....;
data.output(new ElasticSearchOutputFormat<>(config, list, new ElasticsearchSinkFunction<String>() {
@Override
public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
indexer.add(createIndexRequest(element));
}
private IndexRequest createIndexRequest(String element) {
return Requests.indexRequest().index("iteblog").type("info").source(element);
}
}));