利用 Amazon Web Services Athena 处理 CSV 文件中的 JSON 数据

概览

Amazon Web Services Athena 为用户提供了方便的数据访问和处理能力，很多数据分析人员依赖 Athena 来完成日常的数据分析工作。Athena 可以支持多种数据格式的访问和处理，例如 Parquet，ORC，Iceberg，Delta lake，JSON，CSV 等等。然而有些场景的 CSV 文件中某些字段是 JSON 字符串，例如在本文中使用的电商产品数据（非真实数据）既包含一些基本属性如价格、品牌等，也包含扩展属性，这些扩展属性可能因商品的不同而存在不同的结构，因此在数据库中采用一个字段以 JSON 的形式存储。针对这种混合模式如何利用 Athena 做高效和优雅的处理是我们接下来要讨论的话题。

数据的格式和处理

下面是本文使用的数据格式的截图，其中的 extensions 字段是较为复杂的 JSON 格式：

为了更明确的看到 extensions 字段的格式，下面给出一个 extensions 字段的例子。可以看到整个字段是以字符串类型存储的 JSON 结构：

"{""categories"":[""Best Buy Gift Cards"",""Entertainment Gift Cards""],""hierarchicalCategories"":{""lvl0"":""Best Buy Gift Cards"",""lvl1"":""Best Buy Gift Cards > Entertainment Gift Cards""},""rating"":2,""free_shipping"":true,""image"":""https://cdn-demo.algolia.com/bestbuy/1696302_sc.jpg""}"

因为其数据格式是 CSV，所以只能选用支持 CSV 格式的 SerDer 来建表：

CREATE EXTERNAL TABLE IF NOT EXISTS test_data.products_data(
objectID BIGINT,name string,description string, brand string, type string, price float, price_range string, url string, popularity int, extensions string)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.OpenCSVSerde'
WITH SERDEPROPERTIES ("separatorChar" = ",","quoteChar"="\"","escapeChar"="\\")
LOCATION 's3://athena-data-store-bjs/products_data/'
TBLPROPERTIES ("skip.header.line.count"="1")

可以正常查询数据：

然而当进一步查询 extensions 内部的字段时就会遇到报错了, 这也很容易理解，因为 extensions 字段是个字符串类型，不能直接支持 JSON 方式的操作。

而如果将 extensions 字段定义为结构化的方式如下，是否可以呢？可以做个测试，实际上也是不行的。

CREATE EXTERNAL TABLE IF NOT EXISTS test_data.products_data(
objectID BIGINT,name string,description string, brand string, type string, price float, price_range string, url string, popularity int, extensions extensions struct<categories:array<string>,hierarchicalCategories:array<string>>)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.OpenCSVSerde'
WITH SERDEPROPERTIES ("separatorChar" = ",","quoteChar"="\"","escapeChar"="\\")
LOCATION 's3://athena-data-store-bjs/products_data/'
TBLPROPERTIES ("skip.header.line.count"="1")

因为 SerDer 只能在行的级别上定义，不能进一步为某个列去定义支持 JSON 的 SerDer。所以只能沿着上面已经建好的表继续处理。

1. 基础的处理方式——json_extract 函数

因为 Athena 是基于 Presto 引擎构建的，所以 Presto 的函数在 Athena 中也是支持的。对于轻量级使用的场景，一个比较方便的方式是采用 json_extract 函数，这种方式由 SQL 引擎在查询过程中做实时解析。其优点是不需要任何预处理过程，缺点是实时解析会带来更高的资源消耗和更大的延迟，同时对于逻辑较复杂的情况代码不够简洁、单引号中的部分没有语法检查，容易出错。

2. 优化的处理方式

为了达到更好的性能（也意味着更低的成本）和代码的简洁性，可以对数据进行一步预处理，使得数据的结构信息体现在表结构中，同时将数据存储为处理效率更高的 Parquet 格式。

CREATE TABLE "test_data"."products_data_interim_table" WITH (
  format = 'Parquet',  
  external_location = 's3://athena-data-store-bjs/products_data_interim_table/'
) AS
select objectID, name, description, brand, type, price, price_range, url, popularity, cast(json_parse(extensions) as ROW(categories Array<VARCHAR>, image VARCHAR)) as extensions from test_data.products_data

然后查询数据如下（注意转换为结构化类型后索引从 1 开始）：

通过 cast(json_parse(extensions) as ROW(categories Array<VARCHAR>, image VARCHAR)) as extensions 这样一个转换，JSON 字符串中的格式信息被提取到了表定义中，之后的查询就可以采用直接操作 JSON 的方式获取数据。同时数据扫描量也从之前的 877.85 MB 降到了 5.84 MB，这一方面得益于更准确的数据扫描（Parquet 格式），另外在建表的过程中 Athena 也会自动对数据文件进行压缩。需要注意的一点是这种方式为数据建立了中间过程表，当原始数据更新时需要重新建表才能获取最新数据，生产环境中可通过数据处理流水线将这一步自动化执行。在大多数情况下这一步额外的操作都是值得的。

3. 使用 with 关键字

如果实际情况中数据量不是很大，只需要关注语法的简洁性，那么也可以采用一种更简便的方式，采用 with 关键字：

with interim AS 
(select objectID, name, description, brand, type, price, price_range, url, popularity, cast(json_parse(extensions) as ROW(categories Array<VARCHAR>, image VARCHAR)) as extensions from test_data.products_data)
select name, extensions.categories[1] from interim order by "objectid"

总结

以上三种方式都是基于 100 万行相同的数据所做的测试，查询的逻辑也是相同的，即获取商品的唯一标识，名称及第一个所属的分类，在查询语句不同时所得的结果也可能有较大不同。

select objectid, name, extensions.categories[0] as category0 from test_data.products_data_interim_table order by “objectid” 简单对比如下：

上面三种方式都可以很好的处理 CSV 中的 JSON 数据，视场景不同可做灵活选择。

第一种方式的最大优点是不需要对数据做预处理，也不需要定义临时试图，针对轻量级查询较为适合。

第二种方式多了一步数据预处理，需要额外的自动化过程才能确保降低维护成本，但优点也非常明显：1. 后续的数据处理语法简单；2. 大幅降低的数据查询成本，这是因为 Athena 的费用和所扫描的数据量直接相关，具体信息可参考文档。

第三种方式实际上是在做较复杂的数据处理和查询时经常用到的方式，将复杂的转换逻辑放在 with 关键字建立的临时视图中，后续处理逻辑的语法简单很多。但是这种方式基本不会带来性能和成本的优势。

另外，这几种处理方式都是在保留原 JSON 结构的前提下所做的处理，而在某些情况下应用只需要用到 JSON 中某些固定的字段，那么就并不需要保留 JSON 结构的灵活性，针对这种情况建议的方式是使用类似第二种方式构建中间表，提取出所有需要的字段保存为基本类型如 string，integer 等并将表格式存储为 Parquet，ORC 等列式存储，后续的处理完全面向一张由基础类型列组成的表。

本篇作者