Spark Scala如何从CSV文件生成Cassandra UDT列表（按ID分组）

现有包含ID、ID1、ID2、col1、col2、col3、col4字段的CSV文件，需要按ID字段分组，将每个ID对应的唯一(ID1, ID2)对转换为Cassandra UDT格式的列表。示例输入：

ID ID1 ID2 COL1 COL2 COL3 COL4
1   AA  01   A   B   C    D
1   AA  02   A   B   C    D
1   AA  02   B   C   D    E
1   AA  03   A   B   C    D
2   BB  01   A   B   C    D
2   BB  02   A   B   C    D
3   CC  01   A   B   C    D
3   CC  01   B   C   D    E

期望输出：

1,[{ID1:"AA",ID2:"01"},{ID1:"AA",ID2:"02"},{ID1:"AA",ID2:"03"}]
2,[{ID1:"BB",ID2:"01"},{ID1:"BB",ID2:"02"}]
3,[{ID1:"CC",ID2:"01"}]

尝试使用collect_list/collect_set分组字段，但无法得到目标数组格式。

核心思路 #

要实现需求，需分两步处理：

1. 去重：同一ID下的(ID1, ID2)组合可能重复（如示例中ID=1的ID2=02出现两次），需先保留唯一组合；
2. 构造UDT并分组收集：将每个唯一的(ID1, ID2)对构造成符合Cassandra UDT的键值结构，再按ID分组收集为列表。

方法1：使用Spark SQL实现 #

假设已将CSV加载为Spark DataFrame（表名csv_data），执行以下SQL：

-- 第一步：去重，保留每个ID下唯一的ID1+ID2组合
WITH unique_pairs AS (
    SELECT DISTINCT ID, ID1, ID2 
    FROM csv_data
)
-- 第二步：按ID分组，构造UDT结构并收集为列表
SELECT
    ID,
    collect_list(named_struct('ID1', ID1, 'ID2', ID2)) AS udt_list
FROM unique_pairs
GROUP BY ID

• named_struct用于生成与Cassandra UDT对应的键值对结构；
• collect_list负责将同一ID下的UDT元素收集为列表；
• 若需要输出为示例中的字符串格式，可通过concat函数拼接ID和列表，或在数据导出时做格式化处理。

方法2：使用Python Pandas处理 #

如果用Python脚本处理CSV，代码如下：

import pandas as pd

# 读取CSV文件（根据实际分隔符调整sep参数）
df = pd.read_csv('your_file.csv', sep='\s+')

# 提取目标字段并去重
unique_pairs = df[['ID', 'ID1', 'ID2']].drop_duplicates()

# 按ID分组，构造UDT格式的列表
result = unique_pairs.groupby('ID').apply(
    lambda group: [{'ID1': row['ID1'], 'ID2': row['ID2']} for _, row in group.iterrows()]
).reset_index(name='udt_list')

# 输出为期望的字符串格式
for _, row in result.iterrows():
    # 格式化列表为目标字符串形式
    list_str = str(row['udt_list']).replace("'", '"')
    print(f"{row['ID']},{list_str}")

问题排查 #

你之前使用collect_list未得到目标格式，大概率是因为：

• 没有先对(ID1, ID2)组合去重，导致列表包含重复元素；
• 未通过named_struct（或类似方法）构造UDT的键值对结构，直接收集单个字段无法形成目标数组格式。

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