如何基于S3动态路径下的CSV数据创建对应AWS Athena表

前提准备 #

• 提前确认所有CSV文件的结构：同一目录下的CSV需要统一Schema（字段名、字段类型、分隔符、是否有表头），同目录下CSV结构不一致会导致查询报错
• 确认你的AWS账号拥有Athena建表权限、对应S3路径的读权限、Glue数据目录的修改权限

方案1：手动单表创建（适合少量val1-val2组合场景） #

如果你的val1-val2组合数量不多，可以直接编写CREATE EXTERNAL TABLE语句创建单表，以下是示例代码（以val1=partnerA、val2=customerB的output目录下的订单CSV为例）：

CREATE EXTERNAL TABLE IF NOT EXISTS partnerA_customerB_output_order (
  order_id string,
  create_time timestamp,
  amount double,
  user_id string
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES (
  'field.delim' = ',',
  'skip.header.line.count' = '1' -- 如果CSV没有表头可以删除该行配置
)
LOCATION 's3://bucket/partnerA/data/customerB/output/order.csv' -- 单个文件对应写文件路径，同目录下所有CSV结构统一可直接写目录路径
TBLPROPERTIES ('has_encrypted_data'='false');

如果需要创建intermediate_results目录下文件的对应表，修改表名和LOCATION路径即可。

方案2：批量自动创建（适合大量val1-val2组合场景） #

如果val1-val2组合和CSV文件数量较多，可以通过AWS SDK批量遍历路径生成建表语句执行，以Python boto3为例实现流程如下：

1. 遍历S3路径获取参数映射：列举bucket下的所有对象，提取路径中的val1、val2参数，以及文件路径、所属目录（output/intermediate_results）
2. 预设Schema映射：提前把不同CSV文件名对应的字段配置成字典，比如order.csv对应哪些字段、user.csv对应哪些字段
3. 循环执行建表语句：调用Athena接口批量执行生成的建表SQL，示例代码片段如下：

import boto3

athena_client = boto3.client('athena')
# 替换为你的Athena查询结果默认存储路径
ATHENA_OUTPUT_LOC = 's3://your-athena-result-bucket/path/'

def create_athena_table(table_name, schema_sql, s3_location):
    create_sql = f"""
    CREATE EXTERNAL TABLE IF NOT EXISTS {table_name} (
      {schema_sql}
    )
    ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
    WITH SERDEPROPERTIES (
      'field.delim' = ',',
      'skip.header.line.count' = '1'
    )
    LOCATION '{s3_location}'
    """
    response = athena_client.start_query_execution(
        QueryString=create_sql,
        ResultConfiguration={'OutputLocation': ATHENA_OUTPUT_LOC}
    )
    return response['QueryExecutionId']

表名建议按照{val1}_{val2}_{目录类型}_{文件名}规则命名，避免出现重名问题。

可选优化：使用分区表替代多表 #

如果你不需要为每个val1-val2、每个CSV单独建表，可以选择建立统一分区表，把val1、val2作为分区字段，仅需建一张表即可大幅降低维护成本，示例建表语句如下：

CREATE EXTERNAL TABLE IF NOT EXISTS all_result_data (
  -- 这里填写CSV的公共字段，如果不同CSV结构不同可以分开建立对应的分区表
  col1 string,
  col2 int,
  col3 double
)
PARTITIONED BY (partner_name string, customer_name string, data_type string)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
WITH SERDEPROPERTIES (
  'field.delim' = ',',
  'skip.header.line.count' = '1'
)
LOCATION 's3://bucket/'
TBLPROPERTIES ('has_encrypted_data'='false');

建表后执行MSCK REPAIR TABLE all_result_data;即可自动加载所有路径下的分区，不需要手动创建多张表。

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