7

我制作了以下管道：任务管理器 -> SQS -> scraper worker (我的应用程序) -> AWS Firehose -> S3 文件 -> Spark ->(?) Redshift。

我正在尝试解决/改进的一些事情，我很乐意得到指导：

  1. 刮板可能会获得重复的数据，并将它们再次刷新到消防软管，这将导致火花中的重复。我应该在开始计算之前使用 Distinct 函数在火花中解决这个问题吗？
  2. 我不会删除 S3 处理的文件，因此数据会越来越大。这是一个好习惯吗？（将 s3 作为输入数据库）或者我应该处理每个文件并在 spark 完成后将其删除？目前我正在做sc.textFile("s3n://...../*/*/*")- 这将收集我所有的存储桶文件并运行计算。
  3. 要将结果放在 Redshift（或 s3）中 -> 我该如何逐步执行此操作？也就是说，如果 s3 越来越大，redshift 就会有重复的数据......我之前总是要刷新它吗？如何？
4

1 回答 1

0

我以前遇到过这些问题，尽管不是在一个管道中。这是我所做的。

  1. 删除重复项

    一种。我使用BloomFilter来删除本地重复。请注意，文档相对不完整，但您可以轻松保存/加载/联合/相交布隆过滤器对象。您甚至可以reduce对过滤器进行操作。

    湾。如果您将数据直接从 Spark 保存到 RedShift，您可能需要花费一些时间和精力来更新当前批次的 BloomFilter、广播它，然后进行过滤以确保全局没有重复。在我在 RDS 中使用 UNIQUE 约束并忽略该错误之前，但不幸的是 RedShift 不遵守该约束。

  2. 3. 数据越来越大

我使用 EMR 集群运行s3-dist-cp 命令来移动和合并数据（因为通常有很多小日志文件，这会影响 Spark 的性能）。如果您碰巧使用 EMR 来托管 Spark 集群，只需在分析之前添加一个步骤，将数据从一个存储桶移动到另一个存储桶。该步骤将command-runner.jarjar 作为自定义 jar，命令看起来像

s3-dist-cp --src=s3://INPUT_BUCKET/ --dest=s3://OUTPUT_BUCKET_AND_PATH/ --groupBy=".*\.2016-08-(..)T.*" --srcPattern=".*\.2016-08.*" --appendToLastFile --deleteOnSuccess

请注意，原始distcp不支持合并文件。

通常，您应该尽量避免将已处理和未处理的数据放在同一个存储桶（或至少是路径）中。

于 2016-08-03T11:25:08.863 回答