donglei created FLINK-20579:
-------------------------------
Summary: eash es sink will have
Key: FLINK-20579
URL: https://issues.apache.org/jira/browse/FLINK-20579
Project: Flink
Issue Type: Improvement
Reporter: donglei
BulkProcessorListener beforebulk must have the same route to speed up write to
es
As we know bulk with same route will send to es only one node and with one
netio one disk io so every
!http://km.oa.com/files/photos/captures/202007/1593922902_79_w1275_h710.png!
Therefore, we take the following method. The beforeBulk in
ElasticsearchSinkBase writes the same bulk according to the same batch. like
this,
private class BulkProcessorListener implements BulkProcessor.Listener {
@Override
public void beforeBulk(long executionId, BulkRequest request) {
if (routePreBulk) {//Need to verify, whether there is a route set upstream
String routing = UUID.randomUUID() + "_" + executionId;
List<ActionRequest> requests = request.requests();
requests.forEach(x -> {
((IndexRequest) x).routing(routing);
});
LOG.info("start bulk actions: {}, routing: {}", request.numberOfActions(),
routing);
}
}
The advantage of this is that when there are many es fragments later, because
every bulk has the same route sent to the same es node, it saves es data
splitting time and data landing time, and improves es performance.
Preliminary estimates, this part can improve the performance of more than 2
times.
The discussion points here are:
Q: can we use keyby and with same route value
A: Since we use this function to improve performance, setting the same route
value after upstream keyby cannot guarantee that all data will be sent in one
batch, such as 1w data and one route value, but there is no guarantee that 1w
data will be in the same batch. .
Q: How to judge whether to add route value
A: Since oceanus cannot provide an external API interface, it is recommended to
sample here, for example, to see if there is a route in a batch, if there are
none, think that this sink does not need a route value.
Q: Is the data uniform
A: we has been running for a long time. In this setting, because bulk is route
value is uniform, es data is uniform
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
