如何在AWS Glue中将JSON字典键转为表格的ID字段？

}

希望通过AWS Glue ETL任务将其转换为如下格式的表格：

| ID       | attr_a | attr_b |
|----------|--------|--------|
| ID_001   | bla    | blub   |
| ID_002   | foo    | bar    |
| ...      |        |        |

但目前得到的表格格式如下：

| ID_001 | ID_002 | ... |
|--------|--------|-----|

在pandas中使用`df.transpose()`即可解决，但在自定义转换中获取的`DynamicFrameCollection`里遇到了困难。请问在Spark/AWS Glue中有没有首选的转换方法？或者有没有更好的方式将该数据导入Athena表格？

---

# 解决方案

## 一、Spark/AWS Glue ETL转换方法

### 方法1：DynamicFrame转Spark DataFrame处理
将Glue的DynamicFrame转为Spark DataFrame后，利用Spark内置函数完成结构转换，是通用且高效的方案：
```python
from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)

# 读取JSON文件（开启multiline支持单条大JSON）
dynamic_frame = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-bucket/path/to/json/"]},
    format="json",
    format_options={"multiline": True}
)

# 转为Spark DataFrame
df = dynamic_frame.toDF()

# 获取所有顶层ID列名
id_columns = df.columns

# 构造stack表达式，将列维度数据转为行维度
stack_expr = f"stack({len(id_columns)}, " + ", ".join([f"'{col}', {col}.attr_a, {col}.attr_b" for col in id_columns]) + ")"
result_df = df.selectExpr(stack_expr).toDF("ID", "attr_a", "attr_b")

# 可选：转回DynamicFrame继续后续Glue操作
result_dynamic_frame = glueContext.create_dynamic_frame.from_df(result_df, glueContext, "transformed_result")

方法2：Glue原生FlatMap转换 #

直接在DynamicFrame上使用flatMap自定义转换逻辑，手动展开顶层键值对：

from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)

# 读取JSON文件（单条大JSON需开启multiline）
dynamic_frame = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-bucket/path/to/json/"]},
    format="json",
    format_options={"multiline": True}
)

# 自定义展开函数：将单条大记录拆分为多条目标格式记录
def flatten_records(rec):
    flattened_list = []
    for id_key, attrs in rec.items():
        flattened_list.append({
            "ID": id_key,
            "attr_a": attrs.get("attr_a"),
            "attr_b": attrs.get("attr_b")
        })
    return flattened_list

# 执行flatMap转换
result_dynamic_frame = dynamic_frame.flatMap(flatten_records)

二、直接通过Athena导入并转换 #

无需经过Glue ETL，可直接在Athena中创建外部表并完成结构转换：

步骤1：创建临时原始表 #

CREATE EXTERNAL TABLE temp_raw_data (
  data map<string, struct<attr_a:string, attr_b:string>>
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://your-bucket/path/to/json/'
TBLPROPERTIES ('has_encrypted_data'='false', 'serialization.format'='1');

步骤2：创建目标表并插入转换后的数据 #

-- 创建目标表
CREATE EXTERNAL TABLE target_data (
  ID string,
  attr_a string,
  attr_b string
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
LOCATION 's3://your-bucket/path/to/target-data/';

-- 插入转换后的数据，自动展开map中的键值对
INSERT INTO target_data
SELECT 
  map_keys(data)[idx] AS ID,
  map_values(data)[idx].attr_a AS attr_a,
  map_values(data)[idx].attr_b AS attr_b
FROM temp_raw_data
CROSS JOIN UNNEST(SEQUENCE(0, SIZE(map_keys(data))-1)) AS t(idx);

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