👋👋 I have a question about batch mode and S3, I hope this is the right place to ask! Ordinarily, we consume real-time IoT data as messages from kafka using Flink in streaming mode. Typical schema such as

{"timestamp": 12345, "sensor": "sensor-1", "measurement": "temperature":, "value": 99}

If we dump these messages to S3, is there a simple way to reprocess the historical messages using Flink in batch mode? E.g. after deploying a new model version in our pipeline. Ideally, these messages would be partition by time and sensor and stored in parquet format although we are not wedded to this. I’d be particularly interested in reprocessing a subset of the historical messages, e.g. just a particular sensor. Thanks in advance 🙏

Flink's FileSystem connector supports reading from Parquet files as a source, and as it's Hive-compatible I would guess that you might be able to get it to work with S3. You'd have to spin up a Flink cluster specifically for the backfilling purpose, as it probably wouldn't make sense to feed the stream into the real-time cluster for watermarking reasons. Another option would be to have a job that reads the records from S3 and puts them on a Kinesis stream or Kafka topic, and then feed this into Flink to process the backfill job. There's technically no need to do it with Flink in batch mode, as Flink in streaming mode is strictly more powerful (as it can process both bounded and unbounded streams). But if you wanted to automate the Flink cluster provisioning and removal then batch mode would probably be better, using a source connector that supports bounded scans (see https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/overview/). Otherwise you would just need some way to recognize when the backfill process is adequately complete such that you could take down the backfill cluster.

@Lachlan Kermode thank you for your reply! Yes, we'll absolutely be using a separate job to avoid any conflicts with the real-time streams. And thank you for the comments on batch vs streaming

No worries! I'm not sure about the most common pattern. My guess would be that running a Flink cluster in Application mode would be the way to go, but I don't know if this is a pattern others use. As for loading messages effectively from S3, I've had some success with Athena (which makes loading larger data sets from S3 more reasonable in latency terms). In summary, you could do something like: 1. Spin up a Flink cluster in Application and batch mode. 2. Load historic events into a filesystem or Kafka topic using Athena. 3. Start Flink job reading from the source.

Why can't you put Kafka in between, you can read messages from the beginning from a kafka topic.

@Lachlan Kermode thanks again for the input, Athena piping messages to kafka sounds like a good solution. Did you write a custom application for this?

@James Ramsay I haven't done this specifically, I've only worked with Athena and Kinesis/Flink separately in an AWS architecture and can imagine it working reasonably. There might also be a more effective way to plug the messages straight into Flink with Athena that would avoid having to track the backfill 'job' completion (as you could then treat it as a bounded stream)-- i.e. by adapting the Hive source connector so that it works for Athena.

@James Ramsay - what issues are you concerned about re using the Flink FileSource as a bounded source when reading archived data from S3? We do that in Flink workflows, and (aside from bits of missing DataSet-ish functionality) it works fine.

Hi @Ken Krugler - thanks for your input! Whilst we work with unbounded kafka sources a lot, we are relatively inexperienced with batch data and the FileSource My initial concerns with FileSource were: 1. Does it support S3? It sounds like it does... 2. Does it support consuming entire directories and not just specific files? Again, I think it does.. 3. How much control would FileSource sink give us to reprocess only a subset of archived data such as a certain time-range or sensor? I suppose there is no filtering built into the FileSource? But we could control this using a sensible S3 partitioning strategy?

1. Yes 2. Yes 3. For date-stamped directories, in the past I used the older file source API, so I could build a Hadoop input source that only referenced a sub-set of the directories. With the new FileSource API you should be able to do this same thing via a custom

SplitEnumerator

.