amazon-s3 - 如何使用 AWS Glue 将许多 CSV 文件转换为 Parquet

Question

我使用 AWS S3、Glue 和 Athena 进行以下设置:

S3 --> 胶水 --> 雅典娜

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

由于我使用 Athena，我想将 CSV 文件转换为 Parquet。我现在正在使用 AWS Glue 来执行此操作。这是我正在使用的当前流程:

运行 Crawler 以读取 CSV 文件并填充数据目录。

运行 ETL 作业以从数据目录创建 Parquet 文件。

运行爬虫以使用 Parquet 文件填充数据目录。

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

Answer 1

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

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

遍历目录/数据库/表

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

经过一番折腾，我想出了下面的脚本来完成这项工作。我已经使用 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()