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

Question

我将AWS S3，Glue和Athena用于以下设置：

S3->胶水->雅典娜

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

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

1. 运行Crawler以读取CSV文件并填充数据目录。
2. 运行ETL作业以从数据目录创建Parquet文件。
3. 运行搜寻器以使用Parquet文件填充数据目录。

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

Answer 1

我遇到了完全相同的情况，我想有效地循环通过搜寻器分类的目录表，这些目录表指向csv文件，然后将它们转换为镶木地板。不幸的是，网络上尚无可用信息。这就是为什么我在LinkedIn上写了一个博客来解释我是如何做到的。请阅读；特别是第5点。希望能有所帮助。请让我知道您的反馈。

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

5. 遍历目录/数据库/表

作业向导附带了用于在数据源上运行预定义脚本的选项。问题是您可以选择的数据源是目录中的单个表。它没有让您选择在整个数据库或一组表上运行作业。您仍然可以稍后修改脚本，但是在胶粘目录中遍历数据库表的方式也很难找到。有目录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()