将 S3 数据文件夹转换为分桶和分区数据存储

Question

我有一个包含 TB 结构化数据的文件夹，它们都遵循固定模式。文件夹结构如下：

s3://main-bucket/
                folder-1/ <= One folder will only contain data from one algorithm
                                part-0000-1.csv <= Created on 06/01/2019 by algorithm1
                                part-0000-2.csv <= Created on 06/01/2019 by algorithm1
                folder-2/
                                part-1000-1.csv <= Created on 06/02/2019 by algorithm2
                                part-1000-2.csv <= Created on 06/02/2019 by algorithm2

使用 Spark 或 Athena 查询主存储桶效率非常低，我想将其转换为如下格式：

s3://data-bucket/
                algorithm1/ <= derived from a column in folder-1/part files and all rows in this folder has the same algorithm value
                  2019/                   
                    06/
                      01/
                        part-0001.parquet
                algorithm2/
                    2019/                   
                       06/
                         02/
                            part-0001.parquet

无论如何要这样做？

Answer 1

据推测，您的数据文件中有一个字段包含日期元素或至少一个日期字段。

分区文件夹通常包含字段的名称，例如：

algorithm1/year=2019/month=06/day=01/part-0001.parquet

这样，Athena 只需扫描路径名就可以“知道”每个目录中的内容。

转换此格式的最简单方法是使用 Athena 本身。基本上：

• 定义描述并指向现有数据的输入表
• 定义一个配置为 Parquet 的输出表，定义用于分区的字段
• 从输入表中选择数据到输出表中

见：Converting to Columnar Formats - Amazon Athena

Answer 2

由于您的数据中没有日期字段，因此您将无法仅使用 Spark 创建该分区。

一种可能的解决方案包括列出 S3 存储桶中的所有 CSV 文件并收集元数据：creation/modification time 和 owner。为此，您可以使用 Hadoop 文件系统 API。像这样的：

import org.apache.hadoop.fs._

val path = new Path("s3://main-bucket/")
val fs = path.getFileSystem(spark.sessionState.newHadoopConf)
val files = fs.listStatus(path)

val filesMeta = files.map{f => (f.getPath().toString, f.getModificationTime(), f.getOwner())}

对于递归列表，您可以改用listFiles。

使用该列表，您可以创建一个包含列的 DataFrame：file_path, timestamp, owner，您可以使用列 input_file_name() 从所有数据中获取 DataFrame 并添加列 timestamp 和 creator，并添加一些格式日期。

现在您有了列创建者和日期，因此您可以编写 parquet 并按这 2 列进行分区以获得您正在寻找的结构。

Answer 3

如果您的数据文件的名称中包含日期时间，您可以在 Athena 查询中使用“秘密”变量 $PATH。然后，您可以利用 regexp_extract 和 datetime functions 创建列，然后可以在 CTAS 查询中使用这些列来对数据进行分区。

这是我的 S3 中模拟文件的结构：

s3://main-bucket/questions=59541533
├── folder-1
|    ├── file1-2019-01-01.json
|    ├── file1-2019-01-02.json
|    ├── file1-2019-02-01.json
|    └── file1-2019-02-02.json
├── folder-2
|    ├── file1-2019-01-01.json
|    ├── file1-2019-01-02.json
|    ├── file1-2019-02-01.json
|    └──  file1-2019-02-02.json

然后我定义了一个指向folder-1的表：

CREATE EXTERNAL TABLE `stackoverflow`.`question_59541533_v1`(
  `foo` int,
  `bar` int)
ROW FORMAT SERDE
  'org.openx.data.jsonserde.JsonSerDe'
WITH SERDEPROPERTIES (
  'paths'='row,uf')
STORED AS INPUTFORMAT
  'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  's3://main-bucket/questions=59541533/folder-1'
TBLPROPERTIES (
  'classification'='json',
  'compressionType'='none'
  'typeOfData'='file')

这里是查询的样子，因为文件名是确定性的，所以我用相当幼稚的regexp_extract 从文件名中提取日期。

SELECT
    YEAR("file_date") as "year",
    MONTH("file_date") as "month",  
    DAY("file_date") as "day"
FROM(
    SELECT
        DATE(regexp_extract("$PATH", '(.)*/(.)*([0-9]{4}-[0-9]{2}-[0-9]{2}).json', 3)) as "file_date",
    FROM
        "stackoverflow"."question_59541533_v1"
)
ORDER BY month, day

结果：

year  | month | day
---------------------
2019  | 1     | 1
2019  | 1     | 2
2019  | 2     | 1
2019  | 2     | 2

现在，您可以使用 CTAS 查询一次性转换数据并对其进行分区（不要忘记选择所有原始数据）

CREATE TABLE partitioned_and_in_parquet
WITH (
    format = 'PARQUET',
    external_location = 's3://new_s3_location/',
    partitioned_by = ARRAY['year', 'month', 'day']
) AS (
    SELECT
        * , -- select existing data
        YEAR("file_date") as "year",
        MONTH("file_date") as "month",  
        DAY("file_date") as "day"
    FROM(
        SELECT
            * ,  -- select existing data
            DATE(regexp_extract("$PATH", '(.)*/(.)*([0-9]{4}-[0-9]{2}-[0-9]{2}).json', 3)) as "file_date",
        FROM
            "stackoverflow"."question_59541533_v1"
    )    
)

这种方法的优点是您可以对数据进行分区并一次性将其转换为镶木地板。最重要的是，您会提前知道会收取多少费用，因为 Athena 定价是基于被扫描的数据量，即 1TB = 5 美元。

这种方法的一个主要缺点是 CTAS 查询的限制是一次只能写入 100 个分区。因此，如果您有超过 3 个月的数据查询将失败，除非您添加 WHERE 子句来克服这个问题。您可以在WHERE 子句中使用$PATH。我记得我前段时间测试过它，即使文件位于同一个“文件夹”中，Athena 也不会扫描文件的内容，因此，您不会被多次收费。但我建议验证这一点。 或者，只是停在几个月的水平。

您需要考虑的其他事项是 CTAS 查询的输出文件数量及其大小。一般来说，你可以使用bucketing，即bucketed_by = ARRAY['some_column'], bucket_count = 3来控制它，但它可能会增加运行时间。

无论如何，如果您决定与 Athena 一起做这件事，还有很多事情可以玩。