java - 创建n个任务并在Spring Batch中并行执行
问题描述
我有读取 100 个 S3 文件夹的 csv 文件的要求。在单次执行中,它可能仅在少数 S3 文件夹中获取文件,例如 60 个文件夹有文件。我需要处理这 60 个文件并将这些数据发布到 Kafka 主题中。这项工作需要每 4 小时安排一次。CSV 数据可以是小记录,也可以是 6 GB 等海量数据。我必须用 Java 开发并部署到 AWS。考虑使用 Spring Batch: 如下步骤: 1. 遍历所有 100 个 S3 文件夹并识别每个文件夹有文件,例如 60 个文件夹有文件。2. 创建许多作业\任务,例如 60 个作业并并行执行它们。
限制:我不应该在这个过程中使用 AWS EMR。
请建议我用最少的故障数据过程来处理这种最佳性能的好方法。
解决方案
这是您考虑的一种可能的方法。(仅供参考,我已经使用我在此处概述的策略使用 spring-batch 和线程进行了文件处理。但是该代码属于我的公司,无法共享。)我建议您阅读这些文章以了解如何使用 spring 进行扩展-批。
一、spring-batch 文档 https://docs.spring.io/spring-batch/docs/current/reference/html/scalability.html
接下来,来自stackoverflow本身的一篇好文章。 最佳 Spring 批量扩展策略
在阅读并理解所有不同的方式之后,我建议您专注于分区, https://docs.spring.io/spring-batch/docs/current/reference/html/scalability.html#partitioning
这也是我使用的技术。在您的情况下,您可以从分区器中为每个文件生成线程数。
您可能需要维护状态,即文件是否分配给线程。'Processing' 和 'Completed Processing' 也可以是代码中的状态。这取决于您的要求。(我在一个单例中维护了一整套状态,所有线程在拾取文件后都会更新,并完成处理文件等)
您还需要考虑在 4 小时窗口结束之前完成每个文件。您可能能够保持文件原样,或者您可能希望在处理时将其移动到新位置。或在处理时重命名文件。同样,这取决于您的要求。但是您需要考虑这种情况。(在我的情况下,我通过添加一个由时间戳(以毫秒为单位)组成的唯一后缀来重命名文件,因此它不能被新文件覆盖。)
最后,来自博客的示例,该示例通过分区器处理 5 个 csv 文件。您可以从这个示例开始。 https://www.baeldung.com/spring-batch-partitioner
并搜索更多样本,看看这是否是您想要采取的方法。祝你好运。
推荐阅读
- node.js - 如何从用户那里接受 Node js 中的时间
- symfony - Symfony 控制器即服务问题与测试速度
- nginx - 如何在 Nginx 访问日志中记录 grpc 状态?
- javascript - “typeof 导入”类型上不存在属性“render”
- excel - 循环引用和迭代在 Excel 中的工作原理
- qt - 如何在 Qml 的 ListView 中添加搜索选项
- r - 使用闪亮的应用程序更新 Docker 容器内的文件
- python-3.x - 安装自定义 Gym 环境
- python - 如何在python中解决这个通用算法
- iteration - Detectron2中每次迭代多少张图像