flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Mike Urbach <mikeurb...@gmail.com>
Subject Re: Efficient Stateful Processing with Time-Series Data and Enrichments
Date Thu, 24 May 2018 06:57:29 GMT
Hi Sihua,

I will test keying by device ID. I was trying to implement this suggestion:
https://stackoverflow.com/a/49395606, but I guess that may be unnecessary
in my case.

Thanks,

Mike

On Wed, May 23, 2018 at 11:30 PM, sihua zhou <summerleafs@163.com> wrote:

> Hi Mike,
> if I'm not misunderstand, you are doing aggregation for every device on
> the stream. You mentioned that, you want to use the MapState to store the
> state for each device ID? this is a bit confusing to me, I think what you
> need maybe a ValueState. In flink, every keyed state(Value, MapState,...so
> on) is already scoped to the key that you keyed. For example,
> *source.keyBy(deviceId).process(processFunction);*
> if you keyBy the source by deviceId, then in processFunction every keyed
> state is scoped to the deviceID internally, you don't need to use the
> MapState<DeviceID, ?> to maintance the device state yourself.
>
> Concerning to the TTL question. I think the tumbling windows & the
> per-window state is enough for you, than that is a better way to go
> currently.
>
> Best,
> Sihua
> On 05/24/2018 13:46,Mike Urbach<mikeurbach@gmail.com>
> <mikeurbach@gmail.com> wrote:
>
> Hi,
>
> I have a two-part question related to processing and storing large amounts
> of time-series data. The first part is related to the preferred way to keep
> state on the time-series data in an efficient way, and the second part is
> about how to further enrich the processed data and feed it back into the
> state.
>
> For the sake of discussion, let's say that I am tracking tens to hundreds
> of millions of IoT devices. This could grow but that's what I'm looking at
> right now. I will receive an initial event from each device, as well as an
> unknown number of subsequent events. I will need to aggregate together all
> the events related to one device for some period of time after the initial
> event, say 1 hour, at which point I can discard the state. After that, I
> will never hear from that device again. (I'm not actually working with IoT
> devices, but that is the gist. At any given point in time, I will have
> millions of active keys, and as some keys expire new keys are added).
>
> The output of my application should contain the full state for a given
> device, and a new output should be generated every time a new event comes
> in. This application must be fault tolerant. I am currently checkpointing
> my state using the RocksDB state backend.
>
> Part 1 of my question is how best to manage this state. This sounds like
> an excellent use case for State TTL (https://cwiki.apache.org/conf
> luence/display/FLINK/FLIP-25%3A+Support+User+State+TTL+Natively). Since
> this is still a pending feature under active discussion, I did some reading
> about how others have dealt with similar use-cases. What I gleaned boils
> down to this: naively storing everything in one large MapState keyed by
> device ID, and using Triggers to clear the state 1 hour after the initial
> event will lead to far to many Triggers to be efficient.
>
> An alternate approach is to bucket my devices into a far smaller amount of
> keys (not in the millions, maybe thousands), and maintain a MapState for
> each bucket. I can fire a Trigger every minute for every bucket, and
> iterate over the MapState to clear any state that has past its TTL.
>
> A similar, alternative approach is to use tumbling Windows to achieve the
> same effect. Every incoming event has a copy of the timestamp of the
> initial event for that device (think of it as when the device came online),
> so I can use that for event time, and let the watermarks lag by 1 hour. The
> devices are bucketed into some fixed amount of keys like above, so I will
> have a Window for each bucket, for each time slice. The Window has a
> Trigger that eagerly fires and purges each element, and a
> ProcessWindowFunction updates a MapState using per-window state, so that
> when a Window expires I can clear the state. I am currently using this
> approach, since it uses Flink's own Windowing and per-window state to clear
> old data, rather than manually doing it with Triggers.
>
> Other than waiting for the State TTL feature, is there a more efficient
> approach to maintain the aggregate state of all events related to one
> device, and output this every time a new event arrives?
>
> Part 2 of my question relates to how I can enrich the state I have
> accumulated before generating outputs. I have some set of enrichments I'd
> like to do using AsyncFunctions to call out to external services. The issue
> is some enrichments require data that may never be present on any one
> event; I need to work with the stream of aggregated data described above to
> be able to make some of those calls. Furthermore, some enrichments might
> need the data added by other enrichments. I would like to feed the enriched
> data back into the state.
>
> This initially sounded like a perfect use case for an IterativeStream,
> until I tried to use it and realized the only way to enable checkpointing
> was to force it using a feature that is deprecated in Flink 1.4.2. Is that
> approach a dead end? If checkpoints will never be supported for
> IterativeStream, I don't want to explore this route, but it would be nice
> if checkpointed IterativeStreams are on the roadmap, or at least a
> possibility.
>
> Now I'm kind of stumped. The only way I can think of aggregating together
> all the state *before* applying enrichments, and feeding the enriched data
> back into that state *after* the enrichments is to sink the enriched data
> to Kafka or something, and then create a source that reads it back and
> feeds into the operator that keeps the state. That works, but I'd prefer to
> keep all the data flowing within the Flink application if possible. Are
> there other approaches to creating feedback loops that play well with fault
> tolerance and checkpoints?
>
> I appreciate any suggestions related to the two points above.
>
> Thanks,
>
> Mike Urbach
>
>
> --
> Mike Urbach
>
>


-- 
Mike Urbach

Mime
View raw message