Athena 在 S3 上查询数据的替代方案答案

【问题标题】：Alternatives for Athena to query the data on S3Athena 在 S3 上查询数据的替代方案
【发布时间】：2019-12-09 22:16:57
【问题描述】：

我在 S3 上有大约 300 GB 的数据。假设数据如下所示：

## S3://Bucket/Country/Month/Day/1.csv 

S3://Countries/Germany/06/01/1.csv 
S3://Countries/Germany/06/01/2.csv 
S3://Countries/Germany/06/01/3.csv 

S3://Countries/Germany/06/02/1.csv 
S3://Countries/Germany/06/02/2.csv

我们正在对数据进行一些复杂的聚合，并且由于某些国家/地区的数据很大而某些国家/地区的数据很小，因此 AWS EMR 没有意义使用，因为一旦小国完蛋了，资源就被浪费了，而大国还要长期运转。因此，我们决定将 AWS Batch（Docker 容器） 与 Athena 一起使用。每个国家/地区的一项工作需要一天的数据。

现在大约有 1000 个作业一起开始，当它们查询 Athena 以读取数据时，容器因达到 Athena 查询限制而失败>。

因此，我想知道解决此问题的其他可能方法是什么？我是否应该使用 Redshift 集群，将那里的所有数据和所有容器查询加载到 Redshift 集群，因为它们没有查询限制。但它很昂贵，而且需要很长时间才能完成。

另一种选择是读取 EMR 上的数据并在其上使用 Hive 或 Presto 来查询数据，但同样会达到查询限制。

如果有人能提供更好的选择来解决这个问题，那就太好了。

【问题讨论】：

我相信您已经想到了这一点，但如果由于某种原因您还没有想到，请确保您根据国家和日期对表进行分区。对于这个用例，它将大大加快查询速度并降低成本。
@Theo 实际上是的，我也已经问过这个问题。这里是：stackoverflow.com/questions/57287621/… 但没有得到任何回应。问题是我们从其他来源获取这些数据，我们无法控制他们如何写入数据。所以我需要找到其他方法来读取这些数据并将其作为分区表。
@Theo 如果您也能就此提供一些见解，那就太好了。
我会尽量回答这个问题，谢谢你的链接。我没有看到它，因为它没有标记amazon-athena，我也将添加该标记。
@谢谢。非常感谢！。

标签： amazon-web-services amazon-s3 amazon-redshift amazon-emr amazon-athena

【解决方案1】：

您可以为此目的使用红移光谱。是的，它有点贵，但它具有可扩展性并且非常适合执行复杂的聚合。

【讨论】：

复杂聚合与查询无关。我们正在以编程方式进行。我们仅使用 Athena 来读取数据。并且由于 Athena 查询限制，寻找其他方式从 S3 读取数据。
如果大部分复杂性不在 Athena 查询中，那么我在回答中提出的一些建议可能不正确。我假设查询很复杂，并且批处理中运行的作业非常基本。如果您主要使用 Athena 来读取数据，那么根本不使用 Athena 可能更有意义，而是将 CSV 下载到批量运行的代码中？如果有帮助，还请查看 S3 Select。

【解决方案2】：

一种解决方案是不要同时启动所有作业，而是让它们保持在并发限制范围内。我不知道这对于您使用的工具来说是容易还是困难，但是如果您同时向 Athena 抛出所有查询，它永远不会很好地工作。编辑：看起来你应该能够在 Batch 中限制作业，请参阅 AWS batch - how to limit number of concurrent jobs（默认情况下 Athena 允许 25 个并发查询，因此尝试 20 个并发作业以获得安全边际 - 但还要添加重试启动作业的代码的逻辑）。

另一种选择是不将其作为单独的查询进行，而是尝试将所有内容组合成更少的查询，甚至是单个查询 - 通过按国家和日期分组，或者通过生成所有查询并将它们与 @987654323 粘合在一起@。但是，如果不了解有关数据和查询的更多信息，很难说这是否可能。无论如何，您可能都必须对结果进行后处理，如果您只是按有意义的内容进行排序，那么在查询运行后将结果分成必要的部分并不难。

使用 Redshift 可能不是解决方案，因为听起来您每天只这样做一次，而且您不会经常使用集群。 Athena 会是一个更好的选择，你只需要更好地处理限制。

由于我对您的用例了解有限，我认为使用 Lambda 和 Step Functions 会比 Batch 更好。使用 Step Functions，您将拥有一个启动 N 个查询的函数（其中 N 等于您的并发限制，如果您没有要求提高它，则为 25），然后是一个轮询循环（查看示例以了解如何为此）检查已完成的查询，并启动新查询以将正在运行的查询数保持在最大值。当所有查询都运行时，最终函数可以触发您在完成所有操作后需要运行的任何工作流（或者您可以在每次查询后运行该工作流）。

Lambda 和 Step Functions 的好处是您无需为闲置资源付费。使用 Batch，您将为只等待 Athena 完成的资源付费。由于 Athena（与 Redshift 相比）有一个异步 API，因此您可以运行 Lambda 函数 100 毫秒来启动查询，然后每隔几秒（或几分钟）运行 100 毫秒以检查是否已完成，然后再运行 100 毫秒左右来完成向上。它几乎可以保证低于 Lambda 免费层。

【讨论】：

感谢您的回答。限制批处理中的并发作业是有意义的。我们不是每天都进行此处理，而是在所有 7 天中每周进行一次。关于按国家分组，它的作用不大，而且每个国家的数据量每周都在变化，因此找到小国家并找到合并这些国家的策略将是手动工作。
关于 Lambda 和 Step Functions，我还没有研究过，但我也会尝试它们。感谢您指出该选项。
使用 Step Functions 将比 Batch 做更多的工作，因为您必须将流程拆分为多个部分，并描述工作流程等等，但如果您让它工作，您将拥有比 Batch 更强大的解决方案，并且可以免费运行（Athena 查询除外，但它们在任一解决方案中都是相同的）。
是的，该解决方案需要时间来构建。如果我使用带有 Hive 的 redshift 集群或 EMR 集群，克服 Athena 限制会更好吗？由于限制并发作业会大大减慢进程。
Athena 的限制是 20 个并发查询 docs.aws.amazon.com/athena/latest/ug/service-limits.html

【解决方案3】：

据我所知，Redshift Spectrum 和 Athena 的成本相同。您不应该将 Redshift 与 Athena 进行比较，它们有不同的目的。但首先我会考虑解决您的数据倾斜问题。由于您提到了 AWS EMR，我假设您使用 Spark。要处理大小分区，您需要按月份或其他一些等分布的值对数据集重新分区。或者您可以使用月份和国家/地区进行分组。你明白了。

【讨论】：

在比较 Redshift Spectrum 和 Athena 之间的成本时，还要考虑运行 Redshift 集群的成本。它们的每个查询字节的成本可能相同，但 Redshift 需要一个集群，并且 Redshift Spectrum 查询获得的计算资源量与集群的大小（和成本）成正比。
是的，但通常我会保持较小的集群，仅用于重复查询所需的数据。但无论如何，主要问题是分区不均。
我同意可能有一种更有效的方法来解决 OP 的问题，运行大量小查询很少有效率。不过，鉴于手头的信息，很难提出好的建议。似乎他们正在 Athena 查询之后进行大部分处理，这意味着仅重新分区以从分区中删除国家/地区可能意味着必须重写其他所有内容（但这可能是获得更好性能所必需的）。
我需要按每个国家/地区进行处理，因此我无法将多个国家/地区的数据合并在一起。
没有代码示例很难理解。无论如何，如果您按国家/地区分组，这将导致大分区。你有加盟吗？

【解决方案4】：

据我了解，您只需向 AWS Athena 服务发送查询，在所有聚合步骤完成后，您只需从 Athena 保存结果的 S3 存储桶中检索生成的 csv 文件，因此您最终会得到 1000 个文件（每个作业一个） .但问题是并发 Athena 查询的数量而不是总执行时间。

您是否考虑过使用Apache Airflow 来编排和安排您的查询。我将气流视为 Lambda 和 Step Functions 组合的替代方案，但它是完全免费的。它易于在本地和远程机器上设置，具有用于任务监控的 CLI 和 GUI，抽象出所有调度和重试逻辑。 Airflow 甚至有 hooks 与 AWS 服务交互。见鬼，它甚至有一个 dedicated operator 用于向 Athena 发送查询，所以发送查询很简单：

from airflow.models import DAG
from airflow.contrib.operators.aws_athena_operator import AWSAthenaOperator
from datetime import datetime

with DAG(dag_id='simple_athena_query',
         schedule_interval=None,
         start_date=datetime(2019, 5, 21)) as dag:

    run_query = AWSAthenaOperator(
        task_id='run_query',
        query='SELECT * FROM  UNNEST(SEQUENCE(0, 100))',
        output_location='s3://my-bucket/my-path/',
        database='my_database'
    )

我将它用于类似类型的每日/每周任务（使用 CTAS 语句处理数据），这些任务超出了并发查询数量的限制。

有很多博客文章和文档可以帮助您入门。例如：

Medium post：使用 Airflow 自动执行 AWS Athena 查询并在 S3 周围移动结果。
Airflow 安装完整指南，link 1 和 link 2

您甚至可以设置integration with Slack 在查询以成功或失败状态终止时发送通知。

但是，我面临的main drawback 是同时实际执行的查询只有 4-5 个，而所有其他查询都处于空闲状态。

【讨论】：