如何根据条件将字符串数组转换为结构数组

Question

我有一个单列_c0的pyspark数据框。

a|b|c|clm4=1|clm5=3
a|b|c|clm4=9|clm6=60|clm7=23

我正在尝试将其转换为这样的选定列的数据框

clm1,clm2,clm3,clm4,clm6,clm7,clm8
a,    b,   c,   1,  null,null,null
a,    b,   c,   9,   60,  23, null

请注意，我删除了clm5，并添加了clm8。

我正在使用以下代码：

transform_expr = """
    transform(split(_c0, '[|]'), (x, i) -> 
                       struct(
                             IF(x like '%=%', substring_index(x, '=', 1), concat('_c0', i+1)), 
                             substring_index(x, '=', -1)
                             )
            )
    """


    df = df.select("_c0",  explode(map_from_entries(expr(transform_expr))).alias("col_name", "col_value")).groupby("_c0").pivot('col_name').agg(first('col_value')).drop("_c0")  

问题是我有多个巨大的文件，我想在该文件上执行此操作，并且每个文件的结果应包含相同的列（也是一个长列表），如果输入文件中不存在这些列，则它们可以具有空值。如何在上面的代码中添加条件以仅选择列名称列表中存在的那些列？

Answer 1

您可以在列表中包含所需的列，并使用它来过滤转换后的数组：

column_list = ["clm1", "clm2", "clm3", "clm4", "clm6", "clm7", "clm8"]

现在使用filter函数在变换步骤之后添加此过滤器：

filter

这将筛选出列表中不存在的所有列。

最后，使用简单的选择表达式将缺少的列添加为空：

column_filter = ','.join(f"'{c}'" for c in column_list)

transform_expr = f"""
            filter(transform(split(_c0, '[|]'), (x, i) -> 
                               struct(
                                     IF(x like '%=%', substring_index(x, '=', 1), concat('clm', i+1)) as name, 
                                     substring_index(x, '=', -1) as value
                                     )
                    ), x -> x.name in ({column_filter}))
            """