title: Reading data form Elasticsearch into Flink aggregation?
body: <p>I'm trying to update documents in Elasticsearch using Kafka messages (as a StreamSource). Writing to Elasticsearch in bulks using windows and the Elasticsearch connector as a sink is fine, however, we need to update existing data in the documents and read that in a bulk-performant manner (not for every tuple but for e.g. the whole window after a <code>byKey()</code> split that we want to aggregate over)</p>
  
  <p>We are using Storm Trident right now which is performing bulk reads before a <a href="https://storm.apache.org/releases/1.2.2/Trident-state.html" rel="nofollow noreferrer"><code>persistentAggregate</code></a> and writes the updated aggregations back after, minimizing interaction with the backend. I just can't find something similar in Flink - any hints?</p>