将数组拆分为列 pyspark-array 长度变量

Posted 2023-04-13

【中文标题】将数组拆分为列 pyspark-array 长度变量

【英文标题】：Split array into columns pyspark-array length variable

【发布时间】：2020-11-01 01:48:46

【问题描述】：

我在将数组拆分为 pyspark 中的各个列时遇到问题。数组长度是可变的（范围为 0-2064）。我试图重用我找到的一段代码，但由于数据量很大，它不起作用。 这是我将数组转换为字符串（dec_spec_str）后尝试的部分。它需要永远。任何帮助表示赞赏。提前致谢

df2 =df2.select(
        "hashval",
        f.split("dec_spec_str", ",").alias("dec_spec_str"),
        f.posexplode(f.split("dec_spec_str", ",")).alias("pos", "val")
    )\
    .drop("val")\
        .select(
        "hashval",
        f.concat(f.lit("decoded_spec_"),f.col("pos").cast("string")).alias("name"),
        f.expr("dec_spec_str[pos]").alias("val")
    )\
    .groupBy("hashval").pivot("name").agg(f.first("val"))

我还想到了创建大约 4000 个字段的数据框，然后解析并写入该数据框或其副本。如果这样更好，那么请让我知道该怎么做。我仍在接受 pyspark

像这样的输入 decode_spec_str = [1.203.3.455,2.334,1.203,0.345] 像这样输出

【问题讨论】：

您能否提供示例输入和预期输出？这将有助于其他人更好地了解您的问题。

输入是这样的。 decoded_str [1.203.3.455,2.334,1.203,0.345,...] 输出列将如下所示。想象一下，ftr1、ftr2、ftr3.. 等是字段名称，而值来自数组的元素。即使我说得对，格式也不正确 ftr1 ftr2 ftr3 ftr4 ftr5 1.203 3.455 2.334 1.203 0.345

能否请您研究一下解决方案，如果对您有帮助，请帮助接受和投票！

【参考方案1】：

IIUC，这对你来说可能是一个可行的解决方案-

在此处创建 DF

df = spark.createDataFrame([(1,[1.203, 3.455, 2.334, 1.2034, 0.345])],[ "col1","col2"])
#df.show(truncate=False) 
df_grp = df.withColumn("explode_col", F.explode_outer("col2"))
df_grp = df_grp.groupBy("col1").pivot("explode_col").agg(F.avg("explode_col"))
df_grp.show()

输入和处理输出

+----+------------------------------------+
|col1|col2                                |
+----+------------------------------------+
|1   |[1.203, 3.455, 2.334, 1.2034, 0.345]|
+----+------------------------------------+

+----+-----+-----+------+-----+-----+
|col1|0.345|1.203|1.2034|2.334|3.455|
+----+-----+-----+------+-----+-----+
|   1|0.345|1.203|1.2034|2.334|3.455|
+----+-----+-----+------+-----+-----+

现在，你会观察到列名是不是预料之中的，# 在这里对列名重新排序的小技巧

count = 1
for col in df_grp.columns:
  if col != "col1":
    df_grp = df_grp.withColumnRenamed(col, "ftr"+str(count))
    print(col)
    print(count)
    count = count+1

最终输出

+----+-----+-----+------+-----+-----+
|col1| ftr1| ftr2|  ftr3| ftr4| ftr5|
+----+-----+-----+------+-----+-----+
|   1|0.345|1.203|1.2034|2.334|3.455|
+----+-----+-----+------+-----+-----+

【讨论】：

这对我不起作用。数组大小为 4000 和所有。并非所有值都是唯一的。表中的记录数将达到数百万。所以 4000* 200 万。你不能有那么多列，对吧？或者你可以吗？我如何处理每条记录@一次然后添加列并可能连接或联合=得到这个错误Py4JJavaError：调用o16744.pivot时发生错误。 ：org.apache.spark.sql.AnalysisException：数据透视列explode_col 有超过10000 个不同的值，这可能表示错误。如果这是有意的，请设置 spark.sql.pivotMaxValues