How to consume Amazon S3 objects from Flink?

user_13268855 0
Viewed 907

Problem

We have an Apache Flink application which was designed to read events from Kafka and emit the calculated results into ElasticSearch. Because of some resourcing problems we have to fallback from Kafka to Amazon S3.

The messages are published to Amazon S3 buckets in small batches in ndjson format.

The files are organized in the following way: /{year}/{month}/{day}/{hour}.
So, in every hour we create a new folder under which we store the most recent events.

Design

As we have seen Amazon S3 can emit notifications whenever a new object has been created.
We can push these notifications either into an SQS or into a Lambda.

  • As it was stated in this topic SQS is not supported by Flink.
  • In case of Lambda we can get the S3 object and push it on the Kinesis Data Stream

We have also found alternative solutions to avoid writing custom Lambda function:

Question

But in all cases we ended up using KDS. Is there any alternative to push data from Amazon S3 to Flink on object creation?

amazon-kinesis apache-flink amazon-web-services architecture amazon-s3
1 Answers

One solution is to use the readFile method to scan an s3 bucket for new objects. When configured with FileProcessingMode.PROCESS_CONTINUOUSLY and an appropriate polling interval this can work quite well. The key is to define a TextInputFormat with a customized FilePathFilter. The file path filter will recurse through the "directories" in your s3 bucket, and since you've structured them with date parts, the recursing can find new files without scanning through lots of objects in the bucket.

Here is what the custom FilePathFilter will look like. I've been using similar code to discover hundreds of new files every few minutes with this, and it's worked without any issues.

public class S3FilePathFilter extends FilePathFilter {

    Pattern datePartsFromPath = Pattern.compile("\\/(?<year>\\d{4})\\/?(?<month>\\d{2})?\\/?(?<day>\\d{2})?\\/?(?<hour>\\d{2})?");

    private final Duration ageLimit;

    public S3FilePathFilter(Duration ageLimit) {
        this.ageLimit=ageLimit;
    }
    
    @Override
    public boolean filterPath(Path filePath) {

        Matcher matcher = datePartsFromPath.matcher(filePath.toString());

        if (matcher.find()) {
            ZonedDateTime limit = ZonedDateTime.now(ZoneId.of("UTC")).minus(ageLimit);

            int year = NumberUtils.toInt(matcher.group("year"));
            int month = NumberUtils.toInt(matcher.group("month"), limit.getMonthValue());
            int day = NumberUtils.toInt(matcher.group("day"), limit.getDayOfMonth());
            int hour = NumberUtils.toInt(matcher.group("hour"), limit.getHour());

            if (year != limit.getYear()) {
                return year < limit.getYear();
            }

            if (month != limit.getMonthValue()) {
                return month < limit.getMonthValue();
            }

            if (day != limit.getDayOfMonth()) {
                return day < limit.getDayOfMonth();
            }

            return hour < limit.getHour();
        }

        return true;
    }
}
user_11609650
Related