如何使用AWS转换29个共60亿行CSV文件的首列日期格式

推荐优先选用AWS Glue ETL 无服务器作业实现，无需维护集群，可自动扩容匹配TB级数据量处理需求，具体方案如下：

方案1：AWS Glue ETL 作业（首选，适配批量大数据处理） #

• 前置条件：所有CSV文件存放在S3路径下，已创建对应Glue Data Catalog表指向该路径，默认CSV无表头时第一列字段名为_col0，可根据实际表结构调整字段名
• 操作步骤：

1. 进入AWS Glue控制台，创建Spark ETL作业，选择Glue 4.0版本，worker类型选G.1X，初始worker数量设为10，开启自动伸缩，最大worker数设为50（可根据运行速度需求调整）
2. 作业核心PySpark逻辑参考：

from pyspark.sql.functions import col, concat, lit, split, concat_ws
from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

# 读取源CSV文件
df = spark.read.format("csv").option("header", "false").load("s3://你的源桶路径/csv存放目录/")

# 处理第一列格式
processed_df = df.withColumn("_col0", 
    concat(
        # 拆分日期段添加横杠
        split(col("_col0"), "\\|").getItem(0).substr(0,4), lit("-"),
        split(col("_col0"), "\\|").getItem(0).substr(5,2), lit("-"),
        split(col("_col0"), "\\|").getItem(0).substr(7,2),
        lit("|"),
        # 拼接剩余字段内容
        concat_ws("|", split(col("_col0"), "\\|").slice(2, 10))
    )
)

# 输出处理后文件到目标S3路径
processed_df.write.format("csv").option("header", "false").mode("append").save("s3://你的目标桶路径/处理后csv存放目录/")

3. 直接运行作业即可，总数据量580GB左右的场景一般30-60分钟可处理完成，成本约几美元。

方案2：Amazon Athena CTAS 方案（适合熟悉SQL，无代码开发需求的场景） #

不需要写Spark代码，直接用SQL完成转换：

• 先确认源CSV已在Glue Data Catalog中创建对应表，第一列字段名为raw_first_col
• 执行如下SQL完成转换，处理结果会直接存到指定S3路径，同时自动生成目标表：

CREATE TABLE 处理后表名
WITH (
    format = 'CSV',
    external_location = 's3://你的目标桶路径/处理后csv存放目录/',
    field_delimiter = ',',
    skip_header_line_count = 0 -- 源文件有表头的话调整为1
) AS
SELECT
    CONCAT(
        SUBSTR(SPLIT_PART(raw_first_col, '|', 1), 1, 4), '-',
        SUBSTR(SPLIT_PART(raw_first_col, '|', 1), 5, 2), '-',
        SUBSTR(SPLIT_PART(raw_first_col, '|', 1), 7, 2),
        '|',
        ARRAY_JOIN(SLICE(SPLIT(raw_first_col, '|'), 2, CARDINALITY(SPLIT(raw_first_col, '|'))-1), '|')
    ) AS raw_first_col,
    col2, col3, col4 -- 替换为你的实际其他列名，直接原封不动输出即可
FROM 源表名;

• 该方案按扫描数据量收费，580GB扫描量的成本约2.9美元，处理时间15-40分钟。

注意事项 #

• 两种方案均不会修改源文件，处理结果输出到新路径，避免原始数据损坏
• 如果需要输出和原文件数量一致的29个文件，Glue方案可以在写出前添加processed_df = processed_df.coalesce(29)，Athena方案可以输出完成后用S3 Batch Operations合并文件
• 此前Glue DataBrew运行失败大概率是单worker资源不足或并行度配置不够，上述两种方案均自动调度资源，不会出现同类问题

内容的提问来源于stack exchange，提问作者user1761848