本文的基础环境可以参考flink 1.10.1 java版本wordcount演示 (nc + socket),在此基础上增加输出结果到elasticsearch。
1. 添加依赖
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-elasticsearch7_2.11</artifactId>
<version>1.10.1</version>
<!--<scope>${scope}</scope>-->
</dependency>
2. 测试代码
package com.demo.redis;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import java.util.ArrayList;
import java.util.HashMap;
/**
* flink结果写入elasticsearch
*/
public class FlinkESSinkDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<String> dataStream = env.socketTextStream("192.168.0.181",9000);
SingleOutputStreamOperator<String> flatMap = dataStream.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out) throws Exception {
String[] strings = value.split(" ");
for (String s : strings) {
out.collect(s);
}
}
});
SingleOutputStreamOperator<Tuple2<String, Integer>> map = flatMap.map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return Tuple2.of(value, 1);
}
});
SingleOutputStreamOperator<Tuple2<String, Integer>> sum = map.keyBy("f0").sum(1);
DataStream<String> result = sum.map(new MapFunction<Tuple2<String, Integer>, String>() {
@Override
public String map(Tuple2<String, Integer> data) throws Exception {
return data.f0 + ":" + data.f1;
}
});
// 配置es参数
ArrayList<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("localhost", 9200, "http"));
ElasticsearchSink.Builder<String> esBuilder = new ElasticsearchSink
.Builder<String>(httpHosts, new MyEsSinkFunction());
esBuilder.setBulkFlushMaxActions(1);
result.addSink(esBuilder.build());
result.print();
env.execute();
}
public static class MyEsSinkFunction implements ElasticsearchSinkFunction<String>
{
@Override
public void process(String s, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
// 定义写入的数据
HashMap<String, String> dataSouce = new HashMap<>();
String timeStamp = "" + System.currentTimeMillis();
dataSouce.put(timeStamp, s);
// 创建es请求
IndexRequest indexRequest = Requests.indexRequest()
.index("flink-es")
.source(dataSouce);
try {
requestIndexer.add(indexRequest);
}
catch (Exception e)
{
e.printStackTrace();
}
}
}
}
设置只有有记录时,立即写入ES
esBuilder.setBulkFlushMaxActions(1);
3. 启动elasticsearch服务
启动elasticsearch服务以后,使用浏览器访问,出现如下输出,表示elasticsearch服务启动成功。
4. 输出结果
启动程序,在nc输入以下测试字符串,
可以在idea控制台看到输出的数据。
通过浏览器,可以看到写入elasticsearch的数据: