通过 Athena 和/或 Glue 将 S3 数据 ETLing 到 CSV答案

【问题标题】：ETLing S3 data into CSV via Athena and/or Glue通过 Athena 和/或 Glue 将 S3 数据 ETLing 到 CSV
【发布时间】：2019-08-16 19:26:48
【问题描述】：

我有一个充满图像文件的 S3 存储桶 (com.example.myorg.images)，它们都遵循相同的命名约定：

<PRODUCT_ID>_<NUMBER>.jpg

其中<PRODUCT_ID> 是一个长数字（RDS 表中的主键），<NUMBER> 始终是三个值之一：100、200 或 300。例如，存储桶可能包含：

1394203949_100.jpg
1394203949_200.jpg
1394203949_300.jpg
1394203950_100.jpg
1394203950_200.jpg
1394203950_300.jpg
...等

我想编写一个 Athena 或 Glue ETL 进程来查询 S3 存储桶以获取其中的所有图像，并以某种方式将 UNIQUE <PRODUCT_ID> 值提取到表或列表中.

据我了解，Athena 会将该表格/列表备份到可下载的 CSV 文件中；如果为真，那么我将按照我在命令行上需要的方式单独处理该无标题 CSV。

例如，如果上面的 6 张图片是桶中的唯一张图片，那么这个过程将：

查询S3，获取由1394203949和1394203950组成的表/列表
创建一个如下所示的可下载 CSV：

可以是 S3 甚至内存中的文件：

1394203949,1394203950

之前没有使用 Athena 或 Glue 的经验，我正在尝试使用 Athena 查询来完成此操作，但我很难通过树木看到森林。

我在第一部分（S3 查询）的最佳尝试：

CREATE EXTERNAL TABLE IF NOT EXISTS products_with_thumbnails (
  product_id string
) 
ROW FORMAT DELIMITED
  FIELDS TERMINATED BY ','
  ESCAPED BY '\\'
  LINES TERMINATED BY '\n'
LOCATION 's3://com.example.myorg.images/';

这将使用 S3 存储桶中所有内容的文件名设置我的内存表我相信，但是：

如何使此表仅包含唯一的产品 ID（无重复）？
如何仅提取文件名的<PRODUCT_ID> 段（1394203949 而不是1394203949_100.jpg）？

我不偏爱 Athena 或 Glue，并且会很高兴任何解决方案能够满足我的需求。 最坏的情况我可以编写一个 Lambda 来在应用程序层完成所有这些 ETL，但如果有一个类似 Hive 或面向 ETL 的 AWS 服务存在用于执行这类事情，我'宁愿利用它！

提前致谢！

【问题讨论】：

标签： amazon-web-services amazon-s3 etl amazon-athena aws-glue

【解决方案1】：

Athena 查询 inside 文件，而不是文件列表，因此仅使用 Athena 是行不通的（有一些滥用它的方法来实现它，但它们会很昂贵且速度慢，而且不会你想要什么）。

如果图像数量少于十万左右，我认为最好的办法是编写一个或多或少相当于aws s3 ls --recursive s3://some-bucket/ | perl -ne '/(\d+)_d+\.jpg$/ && print "$1\n"' | uniq 的脚本。

如果不止于此，我建议使用 S3 Inventory 和 Athena 进行处理。您可以在此处找到有关如何启用 S3 Inventory 的说明，并使用 Athena 查询库存：https://docs.aws.amazon.com/AmazonS3/latest/dev/storage-inventory.html

使用 S3 Inventory 设置后，您的查询可能如下所示：

SELECT DISTINCT regexp_extract(key, '(\d+)_\d+\.jpg', 1)
FROM the_inventory_table_name

不过，编写处理库存的脚本可能比设置 Athena 表的工作量少。不过，我真的建议使用 S3 Inventory，而不是在有很多对象时直接列出 S3。

【讨论】：

【解决方案2】：

看起来您可以在 S3 中创建 S3 清单的分区文件，按日期分区：

CREATE EXTERNAL TABLE my_inventory(
  `bucket` string,
  key string,
  version_id string,
  is_latest boolean,
  is_delete_marker boolean,
  size bigint,
  last_modified_date timestamp,
  e_tag string,
  storage_class string,
  is_multipart_uploaded boolean,
  replication_status string,
  encryption_status string,
  object_lock_retain_until_date timestamp,
  object_lock_mode string,
  object_lock_legal_hold_status string
  )
  PARTITIONED BY (dt string)
  ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.orc.OrcSerde'
  STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
  OUTPUTFORMAT  'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'
  LOCATION 's3://com.example.myorg.mybucket/com.example.myorg.mybucket/com.example.myorg.mybucket-ORC/hive/';

然后，只要您想查询 my_inventory 表，首先通过为当前日期创建新分区来修复分区文件：

MSCK REPAIR TABLE my_inventory;

最后你可以通过 PrestoDB 的类似 SQL 的语法来查询它：

SELECT key FROM my_inventory WHERE dt <= '<YYYY-MM-DD>-00-00';

其中<YYYY-MM-DD> 是YYYY-MM-DD 格式的当前日期。

然后，您可以将查询结果下载为 CSV 文件并根据需要进行处理。

【讨论】：