如何使用AWS Glue将许多CSV文件转换为镶木地板

Question

我正在使用AWS S3、Glue和Athena进行以下设置：

S3级--

我的原始数据作为CSV文件存储在S3上。我正在使用Glue进行ETL，并使用Athena查询数据。

因为我使用的是雅典娜，所以我想将CSV文件转换为拼花。我现在正在用AWS胶水来做这个。这是我当前使用的流程：

1. 运行爬虫读取CSV文件并填充数据目录

胶水作业一次只允许我转换一个表。如果我有很多CSV文件，这个过程很快就变得无法管理。有没有更好的方法，也许是“正确”的方法，使用AWS Glue或其他一些AWS服务将许多CSV文件转换为Parket？

Answer 1

我不太喜欢胶水，也不喜欢从数据创建模式

以下是如何在雅典娜中做到这一点，它比胶水快得多。

这是针对CSV文件的：

create table foo (
  id int,
  name string,
  some date
)
row format delimited
  fields terminated by ','
location 's3://mybucket/path/to/csvs/'

这是拼花文件：

create table bar 
with (
  external_location = 's3://mybucket/path/to/parquet/',
  format = 'PARQUET'
)
as select * from foo 

即使使用分区，也不需要为拼花地板创建该路径

Answer 2

请参阅编辑以获取更新信息。

S3级--

为什么不直接将CSV格式用于Athena？

https://docs.aws.amazon.com/athena/latest/ug/supported-format.html

CSV是支持的格式之一。为了提高效率，您可以压缩多个CSV文件以加快加载速度。

支持压缩，

https://docs.aws.amazon.com/athena/latest/ug/compression-formats.html

希望有帮助。

编辑：

为什么拼花格式比CSV更有用？

https://dzone.com/articles/how-to-be-a-hero-with-powerful-parquet-google-and

S3级--

有关CSV到拼花地板转换的更多详细信息，

https://aws.amazon.com/blogs/big-data/build-a-data-lake-foundation-with-aws-glue-and-amazon-s3/

Answer 3

我也遇到过同样的情况，我想高效地循环遍历爬虫编目的目录表，这些目录表指向csv文件，然后将它们转换为拼花地板。不幸的是，网络上还没有太多可用的信息。这就是为什么我在LinkedIn上写了一篇博客，解释我是如何做到这一点的。请阅读；特别是第5点。希望这有帮助。请让我知道你的反馈。

注意：根据Antti的反馈，我正在粘贴下面博客中的摘录解决方案：

作业向导提供了在数据源上运行预定义脚本的选项。问题是，您可以选择的数据源是目录中的单个表。它不提供在整个数据库或一组表上运行作业的选项。您可以稍后修改脚本，但在glue catalog中遍历数据库表的方法也很难找到。有目录API，但缺少合适的示例。github示例repo可以通过更多的场景来丰富，以帮助开发人员。

经过一番周旋，我想出了下面的脚本来完成这项工作。我已经使用boto3客户端在表中循环。如果有人需要帮助，我会把它贴在这里。如果你有更好的建议，我也想听听你的意见

import sys
import boto3
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)


client = boto3.client('glue', region_name='ap-southeast-2')

databaseName = 'tpc-ds-csv'
print '\ndatabaseName: ' + databaseName

Tables = client.get_tables(DatabaseName=databaseName)

tableList = Tables['TableList']

for table in tableList:
    tableName = table['Name']
    print '\n-- tableName: ' + tableName

    datasource0 = glueContext.create_dynamic_frame.from_catalog(
        database="tpc-ds-csv", 
        table_name=tableName, 
        transformation_ctx="datasource0"
    )

    datasink4 = glueContext.write_dynamic_frame.from_options(
        frame=datasource0,
        connection_type="s3", 
        connection_options={
            "path": "s3://aws-glue-tpcds-parquet/"+ tableName + "/"
            },
        format="parquet",
        transformation_ctx="datasink4"
    )
job.commit()