Hi Kostas,
thanks for your quick response.
I also thought about using Async IO, I just need to figure out how to
correctly handle parallelism and number of async requests.
However that's probably the way to go..is it possible also to set a number
of retry attempts/backoff when the async request fails (maybe due to a too
busy server)?
For the second part I think it's ok to persist the state into RocksDB or
HDFS, my question is indeed about that: is it safe to start reading (with
another Flink job) from RocksDB or HDFS having an updatable state "pending"
on it? Should I ensure that state updates are not possible until the other
Flink job hasn't finish to read the persisted data?
And another question...I've tried to draft such a processand basically I
have the following code:
DataStream<MyGroupedObj> groupedObj = tuples.keyBy(0)
.flatMap(new RichFlatMapFunction<Tuple4, MyGroupedObj>() {
private transient ValueState<MyGroupedObj> state;
@Override
public void flatMap(Tuple4 t, Collector<MyGroupedObj> out) throws
Exception {
MyGroupedObj current = state.value();
if (current == null) {
current = new MyGroupedObj();
}
....
current.addTuple(t);
...
state.update(current);
out.collect(current);
}
@Override
public void open(Configuration config) {
ValueStateDescriptor<MyGroupedObj> descriptor =
new ValueStateDescriptor<>(
"test",TypeInformation.of(MyGroupedObj.class));
state = getRuntimeContext().getState(descriptor);
}
});
groupedObj.print();
but obviously this way I emit the updated object on every update while,
actually, I just want to persist the ValueState somehow (and make it
available to another job that runs one/moth for example). Is that possible?
On Tue, May 16, 2017 at 5:57 PM, Kostas Kloudas <k.klou...@data-artisans.com
> wrote:
> Hi Flavio,
>
> From what I understand, for the first part you are correct. You can use
> Flink’s internal state to keep your enriched data.
> In fact, if you are also querying an external system to enrich your data,
> it is worth looking at the AsyncIO feature:
>
> https://ci.apache.org/projects/flink/flink-docs-release-1.2/
> dev/stream/asyncio.html
>
> Now for the second part, currently in Flink you cannot iterate over all
> registered keys for which you have state. A pointer
> to look at the may be useful is the queryable state:
>
> https://ci.apache.org/projects/flink/flink-docs-release-1.2/
> dev/stream/queryable_state.html
>
> This is still an experimental feature, but let us know your opinion if you
> use it.
>
> Finally, an alternative would be to keep state in Flink, and periodically
> flush it to an external storage system, which you can
> query at will.
>
> Thanks,
> Kostas
>
>
> On May 16, 2017, at 4:38 PM, Flavio Pompermaier <pomperma...@okkam.it>
> wrote:
>
> Hi to all,
> we're still playing with Flink streaming part in order to see whether it
> can improve our current batch pipeline.
> At the moment, we have a job that translate incoming data (as Row) into
> Tuple4, groups them together by the first field and persist the result to
> disk (using a thrift object). When we need to add tuples to those grouped
> objects we need to read again the persisted data, flat it back to Tuple4,
> union with the new tuples, re-group by key and finally persist.
>
> This is very expansive to do with batch computation while is should pretty
> straightforward to do with streaming (from what I understood): I just need
> to use ListState. Right?
> Then, let's say I need to scan all the data of the stateful computation
> (key and values), in order to do some other computation, I'd like to know:
>
> - how to do that? I.e. create a DataSet/DataSource<Key,Value> from the
> stateful data in the stream
> - is there any problem to access the stateful data without stopping
> incoming data (and thus possible updates to the states)?
>
> Thanks in advance for the support,
> Flavio
>
>
>
--
Flavio Pompermaier
Development Department
OKKAM S.r.l.
Tel. +(39) 0461 1823908 <+39%200461%20182%203908>
