如何从 Flink 使用 Amazon S3 对象？

Question

问题

我们有一个 Apache Flink 应用程序，旨在从 Kafka 读取事件并将计算结果发送到 ElasticSearch。由于一些资源问题，我们不得不从 Kafka 回退到 Amazon S3。

消息以ndjson 格式小批量发布到 Amazon S3 存储桶。

文件按以下方式组织：/{year}/{month}/{day}/{hour}.
因此，我们每小时都会创建一个新文件夹，在该文件夹下存储最新事件。

设计

正如我们所见，每当创建新对象时，Amazon S3 都会发出 notifications。
我们可以将这些通知推送到 SQS 或 Lambda。

• 正如this topic 中所述，Flink 不支持 SQS。
• 对于 Lambda，我们可以获取 S3 对象并将其推送到 Kinesis Data Stream

我们还找到了避免编写自定义 Lambda 函数的替代解决方案：

• Via AWS Data Migration Service
• Via Athena

问题

但在所有情况下，我们最终都使用了 KDS。在创建对象时，是否可以将数据从 Amazon S3 推送到 Flink？

Answer 1

一种解决方案是使用 readFile 方法扫描 s3 存储桶以查找新对象。当配置FileProcessingMode.PROCESS_CONTINUOUSLY 和适当的轮询间隔时，这可以很好地工作。关键是用自定义的FilePathFilter 定义TextInputFormat。文件路径过滤器将递归通过您的 s3 存储桶中的“目录”，并且由于您已使用日期部分对它们进行结构化，因此递归可以找到新文件，而无需扫描存储桶中的大量对象。

这是自定义 FilePathFilter 的外观。我一直在使用类似的代码每隔几分钟就发现数百个新文件，并且它可以正常工作。

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;
    }
}