Hello,
I have a Flink pipeline reading data from Kafka which is keyed (in the
pseudocode example below it is keyed on the first letter in the string, in our
pipeline it is keyed on a predefined key) and processed in sliding windows with
a duration of 60m every 10:th minute. The time setting is eventTime and the
windows processes the data when the window should fire, there are no
incremental processing of the windowed data.
When running with a Heap backend the state behaves “normally”, i.e it stays
within the data size that is expected when the windows have buffered the data
(~200 Mb for this application) and is bounded to around this size independent
of the lifetime of the processing pipeline. However, if the state backend is
changed to the RocksDb backend the states starts to grow indefinitely (is our
observation, we haven’t seen it stop growing at least) to 70-80 Gb in just
above of a month runtime.
I made a save point of the state and downloaded it and analysed which shown
that the state consisted of data from the whole lifetime of the pipeline, of
about equal size for each day. I interpret this as the state has accumulated
the old data which should’ve been deleted during the clearing of the windows.
It is worth noting that the state consists of the input Strings only, so it
should have nothing to do with the histogram calculation?
I have tried to reduce the level0_num_of_files_compation_trigger to be 1 and
the base file size as well as the target file size to trigger more compactions
in the hope of that the compactions would remove the obsolete data which
rendered in no improvement at all (it basically got worse).
Please see the pseudocode below for a code example. The pipeline is more
complex than this and is running on other classes than String input and a
“histogram” output class. Do you have any input or ideas how the state could be
manageable in the Heap case but totally unhandleable during the RocksDb version?
Best regards,
William
class Histogram extends WindowFunction[String, Histogram, TimeWindow] {
def process (key : T, window: TimeWindow, input : Itrable[String]) = {
//Calculate the histogram
}
override def apply(key : T, window: TimeWindow, input : Iterable[String], out:
Collector[Histogram]) : Unit = {
out.collect(process(key, window, input))
}
}
env.getCheckpointConfig.setCheckpointTimeout(400000)
env.getCheckpointConfig.setMinPauseBetweenCheckpoint(450000)
val stateBackend : StateBackend = new RocksDBStateBackend(s3://.., true)
env.setStateBackend(stateBackend)
env.enableCheckpointing(900000)
DataStream<String> stream = env
.addSource(new FlinkKafkaConsumer08<>("topic", new
SimpleStringSchema(), properties));
env.
stream.keyBy(e => e.charAt(0)).timeWindow(minutes(60), minutes(10)).apply(new
Histogram()).name(“Pseudocode").uuid(“Psuedocode”)
William Jonsson
Systems Engineer
Fleet Perception for Maintenance
[cid:nd_logo_d020cb30-0d08-4da8-8390-474f0e5447c8.png]
NIRA Dynamics AB
Wallenbergs Gata 4
583 30 Linköping
Sweden Mobile: +46 722 178 247
william.jons...@niradynamics.se
www.niradynamics.se
Together for smarter safety
