使用 Amazon Athena 分析 S3 中的数据

Neil Mukerje 是 Amazon Web Services 的一名解决方案架构师
Abhishek Sinha 是一名 Amazon Athena 高级产品经理

Amazon Athena 是一种交互式查询服务，可轻松使用标准 SQL 直接分析来自 Amazon S3 的数据。Athena 采用无服务器架构，因此无需设置或管理基础设施，即可立即开始分析数据。您甚至无需将数据加载到 Athena，也没有复杂的 ETL 过程。 Athena 可以直接处理存储在 S3 中的数据。Athena 使用 Presto（一种分布式 SQL 引擎）来运行查询。它还使用 Apache Hive 来创建、放置和修改表和分区。您可以在 Athena 查询编辑器中编写符合 Hive 规范的 DDL 语句以及 ANSI SQL 语句。您还可以在 Athena 上使用复杂的联结、窗口函数和复杂的数据类型。Athena 使用一种称为读时模式(schema-on-read) 的方法，这样您可以在执行查询时将 schema 投射到您的数据上，从而不再需要任何数据加载或 ETL 工作。

Athena 会根据每个查询扫描的数据量向您收费。对数据进行分区、压缩数据或将其转换为 Apache Parquet 等列格式，可以节省成本并获得更好的性能。有关更多信息，请参阅 Athena 定价。

在本博文中，我们演示了如何使用 Athena 来处理来自 Elastic Load Balancer 的日志（预先定义好的文本格式）。我们将演示如何创建表，按照 Athena 使用的格式将数据分区，然后转换为 Parquet 并比较查询性能。

在此例中，原始日志以如下格式存储在 Amazon S3 中。另外有一个独立的年月日前缀，共含 2570 个对象，数据大小为 1TB。

o_athena_1

创建表

如果您熟悉 Apache Hive，那么您可能会发现在 Athena 上创建表一点也不陌生。您可以通过在查询编辑器上编写 DDL 语句的方式来创建表，也可以通过使用向导或 JDBC 驱动程序来创建表。复制以下 DDL 语句并粘贴到 Athena 查询编辑器中以创建表。

CREATE EXTERNAL TABLE IF NOT EXISTS elb_logs_raw_native (
  request_timestamp string,
  elb_name string,
  request_ip string,
  request_port int,
  backend_ip string,
  backend_port int,
  request_processing_time double,
  backend_processing_time double,
  client_response_time double,
  elb_response_code string,
  backend_response_code string,
  received_bytes bigint,
  sent_bytes bigint,
  request_verb string,
  url string,
  protocol string,
  user_agent string,
  ssl_cipher string,
  ssl_protocol string ) 
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
         'serialization.format' = '1','input.regex' = '([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*)[:\-]([0-9]*) ([-.0-9]*) ([-.0-9]*) ([-.0-9]*) (|[-0-9]*) (-|[-0-9]*) ([-0-9]*) ([-0-9]*) \\\"([^ ]*) ([^ ]*) (- |[^ ]*)\\\" (\"[^\"]*\") ([A-Z0-9-]+) ([A-Za-z0-9.-]*)$' ) 
LOCATION 's3://athena-examples/elb/raw/';

请注意“CREATE TABLE”语句中指定的正则表达式。您可以指定任何正则表达式，该表达式会告诉 Athena 如何解读文本中的每一行。您还可以使用 Athena 来查询 JSON 等其他数据格式。处理 CSV、TSV 或 JSON 格式时不需要正则表达式。语句执行成功后，表和 schema 将在数据目录（左侧窗格）中显示。 Athena 使用一种内部数据目录来存储有关表、数据库和分区的信息。它具有极高的持久性，无需任何管理。您可以使用 DDL 查询或通过控制台与此目录进行交互。

o_athena_2

您已经为 Amazon S3 中存储的数据创建了一个表，现在可以查询数据。请注意表 elb_logs_raw_native 指向前缀 s3://athena-examples/elb/raw/。因此，当您在该前缀下添加更多数据（例如新月份的数据）时，表会自动增长。运行一个简单的查询：

SELECT * FROM elb_logs_raw_native WHERE elb_response_code = '200' LIMIT 100;

o_athena_3_1

现在您可以查询所有日志，无需设置任何基础设施或执行 ETL。

数据分区

客户往往会以时间序列格式来存储数据，需要查询某天、某月或某年的特定项目。如果不进行分区，Athena 在执行查询时会扫描整个表。进行分区后，您可以将 Athena 限定为查询特定的分区，从而减少扫描的数据量，降低成本，提高性能。

Athena 使用 Apache Hive 式的数据分区。您可以按多个维度将数据分区，例如按月、周、日、小时或客户 ID 进行分区，也可将数据全部汇聚在一起。

要使用分区，您首先需要更改 schema 定义以包含分区，然后在 Athena 中加载分区元数据。使用同样的 CREATE TABLE 语句，但这次要启用分区。

CREATE EXTERNAL TABLE IF NOT EXISTS elb_logs_raw_native_part (
  request_timestamp string,
  elb_name string,
  request_ip string,
  request_port int,
  backend_ip string,
  backend_port int,
  request_processing_time double,
  backend_processing_time double,
  client_response_time double,
  elb_response_code string,
  backend_response_code string,
  received_bytes bigint,
  sent_bytes bigint,
  request_verb string,
  url string,
  protocol string,
  user_agent string,
  ssl_cipher string,
  ssl_protocol string ) 
PARTITIONED BY(year string, month string, day string)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
         'serialization.format' = '1','input.regex' = '([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*)[:\-]([0-9]*) ([-.0-9]*) ([-.0-9]*) ([-.0-9]*) (|[-0-9]*) (-|[-0-9]*) ([-0-9]*) ([-0-9]*) \\\"([^ ]*) ([^ ]*) (- |[^ ]*)\\\" (\"[^\"]*\") ([A-Z0-9-]+) ([A-Za-z0-9.-]*)$' )
LOCATION 's3://athena-examples/elb/raw/';

o_athena_4

请注意“CREATE TABLE”语句中的“PARTITIONED BY”分句。此数据将按年、月、日分区。在结果部分，Athena 会提醒您加载分区后的表的分区。

借助 ALTER TABLE ADD PARTITION 语句可以加载与某个分区有关的元数据。例如要加载来自 s3://athena-examples/elb/raw/2015/01/01/ 存储桶的数据，您可以运行以下语句：

ALTER TABLE elb_logs_raw_native_part ADD PARTITION (year='2015',month='01',day='01') location 's3://athena-examples/elb/raw/2015/01/01/'
show partitions elb_logs_raw_native_part

o_athena_5_1

现在您可以通过在“WHERE”分句中指定分区来限制每次查询的范围。这时，Athena 扫描的数据范围会减少，完成的速度会更快。示例如下：

SELECT distinct(elb_response_code),
         count(url)
FROM elb_logs_raw_native_part
WHERE year='2015'
        AND month= '01'
        AND day='01'
GROUP BY  elb_response_code

o_athena_6_1

如果您有大量的分区，则手动指定分区可能较为困难。您可以使用 JDBC 驱动程序来自动化执行此过程。如果您的数据已经为 Hive 分区格式，则无需这样做。

将数据转换为列格式

Athena 允许您使用 Apache Parquet 和 Apache ORC 等列格式。通过将数据转换为列格式，不仅有利于提高查询的性能，而且可以节省成本。

将数据转换为列格式的方法有多种。在本博文中，您将利用一种 PySpark 脚本，它大约有 20 行，在 Amazon EMR 上运行以将数据转换为 Apache Parquet。此脚本还会按年、月、日对数据进行分区。在本文发表时，在 US-east 区域的一个双节点 r3.x8large 集群能够以 5 USD 的总成本将 1TB 的日志文件转换为 130GB 的压缩 Apache Parquet 文件（压缩率 87%）。

这时文件在 Amazon S3 中的布局如下：

o_athena_7

请注意文件的布局。此分区格式用键=值的格式指定，会在 Athena 中自动识别为分区。这样您可以使用 msck repair table <tablename> 命令自动加载全部分区。这与 Hive 理解分区数据的方式类似。如果数据并非采用上文所述的键值格式，则需要按前文所述手动加载分区。

创建一个有关该 Parquet 数据集的表。请注意您的 schema 仍然相同，并且您将使用 Snappy 来压缩文件。

CREATE EXTERNAL TABLE IF NOT EXISTS elb_logs_pq (
  request_timestamp string,
  elb_name string,
  request_ip string,
  request_port int,
  backend_ip string,
  backend_port int,
  request_processing_time double,
  backend_processing_time double,
  client_response_time double,
  elb_response_code string,
  backend_response_code string,
  received_bytes bigint,
  sent_bytes bigint,
  request_verb string,
  url string,
  protocol string,
  user_agent string,
  ssl_cipher string,
  ssl_protocol string )
PARTITIONED BY(year int, month int, day int) 
STORED AS PARQUET
LOCATION 's3://athena-examples/elb/parquet/'
tblproperties ("parquet.compress"="SNAPPY");

o_athena_8

要允许目录识别所有的分区，请运行 msck repair table elb_logs_pq 命令。查询完成后，您可以列出全部分区。

msck repair table elb_logs_pq
show partitions elb_logs_pq

o_athena_9_1

性能比较

您可以比较对文本文件和 Parquet 文件执行同样的查询时的性能。

SELECT elb_name,
       uptime,
       downtime,
       cast(downtime as DOUBLE)/cast(uptime as DOUBLE) uptime_downtime_ratio
FROM 
    (SELECT elb_name,
        sum(case elb_response_code
        WHEN '200' THEN
        1
        ELSE 0 end) AS uptime, sum(case elb_response_code
        WHEN '404' THEN
        1
        ELSE 0 end) AS downtime
    FROM elb_logs_pq
    GROUP BY  elb_name)

对经过压缩、分区和分列的数据进行查询

o_athena_10_1

对原始文本文件进行查询

SELECT elb_name,
       uptime,
       downtime,
       cast(downtime as DOUBLE)/cast(uptime as DOUBLE) uptime_downtime_ratio
FROM 
    (SELECT elb_name,
        sum(case elb_response_code
        WHEN '200' THEN
        1
        ELSE 0 end) AS uptime, sum(case elb_response_code
        WHEN '404' THEN
        1
        ELSE 0 end) AS downtime
    FROM elb_logs_raw_native
    GROUP BY  elb_name)

o_athena_11_1

Athena 会根据每个查询扫描的数据量向您收费。通过将数据转化为列格式，并进行压缩和分区，不仅可以节省成本，还可以提高性能。下表比较了通过将数据转换为列格式后实现的节省。

数据集	在 Amazon S3 中的大小	查询运行时间	扫描的数据量	成本
以文本文件存储的数据	1TB	236 秒	1.15TB	5.75 USD
以 Apache Parquet 格式存储的数据*	130GB	6.78 秒	2.51GB	0.013 USD
节省/加速	使用 Parquet 后节省 87%	速度提高 34 倍	扫描的数据量减少 99%	节省 99.7%

（*使用 Snappy 压缩格式压缩）

小结

通过 Amazon Athena 可以使用标准 SQL 来分析 S3 的数据，无需管理任何基础设施。您还可以使用 JDBC 驱动程序，通过商业智能工具来访问 Athena。Athena 会根据每个查询扫描的数据量向您收费。从本博文中可以看出，将数据转换为开源格式不仅可以节省成本，还可以提高性能。

您可以在美国东部（弗吉尼亚北部）和美国西部 2（俄勒冈）区域试用 Amazon Athena。要了解更多信息，请参阅 Amazon Athena 产品页面或 Amazon Athena 用户指南。

亚马逊AWS官方博客