package com.shi.kafkaTest;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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.streaming.connectors.elasticsearch7.RestClientFactory;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.RestClientBuilder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
/**
* @author shiye
* @create 2023-02-20 17:06
*/
public class KafkSourceEsSinkTest {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>("sungrow_cdc", new SimpleStringSchema(), getKafkaPropertes()));
ArrayList<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("10.0.81.167", 9201, "http"));
httpHosts.add(new HttpHost("10.0.81.168", 9201, "http"));
httpHosts.add(new HttpHost("10.0.81.169", 9201, "http"));
ElasticsearchSink.Builder<String> esBuilder = new ElasticsearchSink.Builder<>(httpHosts, getEsSinkFunction());
//刷新前最大缓存的操作数。
esBuilder.setBulkFlushMaxActions(1);
//刷新前最大缓存的数据量(以兆字节为单位)。
esBuilder.setBulkFlushMaxSizeMb(100);
//刷新的时间间隔(不论缓存操作的数量或大小如何)
esBuilder.setBulkFlushInterval(10000);
//设置用户名密码
esBuilder.setRestClientFactory(getRestClientFactory());
stream.addSink(esBuilder.build());
env.execute();
}
//定义kafka配置信息
public static Properties getKafkaPropertes() {
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "10.0.81.158:9092,10.0.81.159:9092,10.0.81.160:9092");
properties.setProperty("group.id", "sungrow_cdc_shiye_test_group4");
properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("auto.offset.reset", "earliest");
return properties;
}
/**
* 定义ES Sink算子
*
* @return
*/
public static ElasticsearchSinkFunction<String> getEsSinkFunction() {
ElasticsearchSinkFunction<String> esSinkFunction = new ElasticsearchSinkFunction<String>() {
@Override
public void process(String event, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
System.out.println("------> " + event);
JSONObject jsonObject = JSONObject.parseObject(event);
// Map<String, Object> map = new HashMap<>();
// map.put("dbName", jsonObject.get("dbName"));
// map.put("tableName", jsonObject.get("tableName"));
// map.put("operation", jsonObject.get("operation"));
IndexRequest request = Requests.indexRequest()
.index("temp_cdc_test")
.source(jsonObject);
requestIndexer.add(request);
}
};
return esSinkFunction;
}
/**
* es 用户名密码
*/
public static RestClientFactory getRestClientFactory() {
RestClientFactory restClientFactory = new RestClientFactory() {
@Override
public void configureRestClientBuilder(RestClientBuilder restClientBuilder) {
CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("elastic", "123456"));
restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
@Override
public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
httpAsyncClientBuilder.disableAuthCaching();
return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
}
});
}
};
return restClientFactory;
}
}