Change elastic bulk strategy
@@ -61,7 +61,8 @@ public class ElasticSearchProgram extends FlinkDataSinkRuntime<ElasticSearchPara
Map<String, String> userConfig = new HashMap<>();
// This instructs the sink to emit after every element, otherwise they would be buffered
- userConfig.put(ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "1");
+// userConfig.put(ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS, "1");
+ userConfig.put(ElasticsearchSink.CONFIG_KEY_BULK_FLUSH_INTERVAL_MS, "5000");
convertedStream[0].flatMap(new FlatMapFunction<Map<String, Object>, Map<String, Object>>() {
