Spark RDD数据格式转换：移除多余括号并转为整数

问题背景 #

需要将给定的Spark RDD数据转换为指定格式：(node_id, (score, edges))，其中edges需移除多余嵌套括号并转为整数类型（例如节点5应输出为(5, (0.09090909090909091, 4, 2, 6))）。

当前输入数据 #

[('4', ('1', '2')),
('10', ('5',)),
('3', ('2',)),
('6', ('2', '5')),
('7', ('2', '5')),
('1', None),
('8', ('2', '5')),
('9', ('2', '5')),
('2', ('3',)),
('5', ('4', '2', '6')),
('11', ('5',))]

当前转换函数 #

def adjDang(line, tc):
    node, edges = line
    print(f'node {node} edges {edges}')
    if edges == None:
        return (int(node),(0,0))
    else:
        if len(edges) == 1:
            newedges = (edges[0]) #remove the comma which is unnecessary check '11'
        else:
            newedges = ()
            for i in range(len(edges)):
                newedges += edges[i]

        print(f'node {node} edge{newedges}')

        return(int(node), (1/tc, newedges))

当前输出结果 #

[(4, (0.09090909090909091, ('1', '2'))),
 (10, (0.09090909090909091, '5')),
 (3, (0.09090909090909091, '2')),
 (6, (0.09090909090909091, ('2', '5'))),
 (7, (0.09090909090909091, ('2', '5'))),
 (1, (0, 0)),
 (8, (0.09090909090909091, ('2', '5'))),
 (9, (0.09090909090909091, ('2', '5'))),
 (2, (0.09090909090909091, '3')),
 (5, (0.09090909090909091, ('4', '2', '6'))),
 (11, (0.09090909090909091, '5'))]

期望输出格式 #

edges部分需移除嵌套括号，所有边值转为整数，示例如下：

[(4, (0.09090909090909091, 1, 2)),
 (10, (0.09090909090909091, 5)),
 (3, (0.09090909090909091, 2)),
 (6, (0.09090909090909091, 2, 5)),
 (7, (0.09090909090909091, 2, 5)),
 (1, (0, 0)),
 (8, (0.09090909090909091, 2, 5)),
 (9, (0.09090909090909091, 2, 5)),
 (2, (0.09090909090909091, 3)),
 (5, (0.09090909090909091, 4, 2, 6)),
 (11, (0.09090909090909091, 5))]

实现思路 #

• 统一处理边数据：无论edges长度为1还是多个，先将每个元素转为整数，生成整数列表
• 构造目标元组：利用Python解包运算符*，将score与整数边列表展开合并为一个元组，消除嵌套结构
• 简化分支逻辑：去掉原代码中对edges长度的判断，用统一逻辑处理所有非None的edges情况

修改后的代码示例 #

def adjDang(line, tc):
    node, edges = line
    node_id = int(node)
    if edges is None:
        return (node_id, (0, 0))
    # 将所有边转换为整数
    int_edges = [int(edge) for edge in edges]
    # 构造包含score和展开边的元组
    result_tuple = (1/tc, *int_edges)
    return (node_id, result_tuple)

代码说明 #

1. 节点ID转换：直接将字符串类型的node转为整数，确保输出节点ID为数值类型
2. None值处理：保留原逻辑，返回(node_id, (0,0))
3. 边类型转换：用列表推导式批量将edges中的字符串元素转为整数，无需区分长度
4. 元组构造：通过*运算符将整数边列表展开，与score合并为一个扁平元组，彻底解决嵌套括号问题

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