【问题标题】：Replace elements in an array by Position in Dataframe - Pyspark用数据框中的位置替换数组中的元素 - Pyspark
【发布时间】：2023-03-24 19:50:01
【问题描述】：

我有一个数据框：

|ID|CTA|
|------|
|11|1  |
|11|2  |
|11|7  |
|45|7  |

我需要按 ID 分组，并且每个 ID 的 ARRAY 长度为 7，但在 CTA 中它有位置时指示为 1

所以我的输出数据框应该是这样的：

|ID|CTAS             |
|------------------- |
|11|[1,1,0,0,0,0,1]  |
|45|[0,0,0,0,0,0,1]  |

你能帮帮我吗？

更新：

如何在数组中保留零？

【问题讨论】：

到目前为止你尝试了什么？
当CUENTA 重复时，例如在2234 客户中，输出应该是什么？
@Kafels 没有帐户重复。
但是客户2234 重复值1。在这种情况下，输出应该是[1, 0, 0, 0, 0, 0, 0]?
@Kafels aaah 好的，因为数据已经处理过了，所以重复了。在这种情况下，客户 2234 有 ACCOUNT [1 和 2]，因此数组应显示 [1, 1, 0, 0, 0, 0, 0]

标签： python dataframe apache-spark pyspark apache-spark-sql

【解决方案1】：

您可以应用TRANSFORM 表达式并遍历sequence(1, 7) 以检查该值是否包含在CTAS 列中：

import pyspark.sql.functions as f

group_df = (df
            .groupBy('ID')
            .agg(f.collect_list('CTA').alias('CTAS')))
# +---+---------+
# |ID |CTAS     |
# +---+---------+
# |11 |[1, 2, 7]|
# |45 |[7]      |
# +---+---------+

pos_df = (group_df
          .withColumn('CTAS', 
                      f.expr('transform(sequence(1, 7), value -> cast(array_contains(CTAS, value) as int))')))
pos_df.sort('ID').show(truncate=False)
# +---+---------------------+
# |ID |CTAS                 |
# +---+---------------------+
# |11 |[1, 1, 0, 0, 0, 0, 1]|
# |45 |[0, 0, 0, 0, 0, 0, 1]|
# +---+---------------------+

【讨论】：

感谢您的帮助！在这种情况下，当运行节目时，零不会显示为 -> 2234 | [1, 1]
请更新您的问题，为此场景提供示例和所需的输出
@JosefinaAndreaArayaTapia 再检查一次，我正在硬编码值以从 1 迭代到 7。

【解决方案2】：

每个 ID 都有其长度为 7 的 ARRAY

只是添加另一种方式（我个人喜欢上面发布的this 答案）。对spark.range 使用不同的交叉连接和左连接来填充缺失的行，然后进行分组并收集：

out = (df.withColumn("CTAS",F.lit(1)).join(
  df.select("ID").dropDuplicates().crossJoin(spark.range(1,8)
                                        .withColumnRenamed("id","CTA"))
       ,on=["ID","CTA"],how='outer').fillna({"CTAS":0})
       .groupBy("ID").agg(F.collect_list("CTAS").alias("CTAS")))

out.show(truncate=False)

+---+---------------------+
|ID |CTAS                 |
+---+---------------------+
|11 |[1, 1, 0, 0, 0, 0, 1]|
|45 |[0, 0, 0, 0, 0, 0, 1]|
+---+---------------------+

【讨论】：

【解决方案3】：

另一种方法可以是自定义函数，以满足您要应用的逻辑。

但是这种方法可能会有点复杂，因为它在实施时几乎没有需要注意的细节。

以上两个答案都更好更快，因为它们使用 Spark Native Functions

这只是使用applyInPandas 的另一种方法

ApplyInPandas 方法

自定义函数 - 包含自定义逻辑

def generate_ctas_list(inp,base_column=None,target_column=None,column_lst=None):
    
    ### Sort the values for your Target Columns 
    tgt_col_values = np.sort(inp[target_column].values)
    
    ### Getting the Max Value , to use to generate the List 
    max_tgt = tgt_col_values[-1]
    
    zeros_lst = np.zeros(max_tgt)
    
    ### Substract 1 to ensure indexing starts from 0
    tgt_col_values = tgt_col_values - 1
    
    ### Update the Zeros List with tgt_col_values to 1 
    zeros_lst[tgt_col_values] = 1    
    
    
    ### Get the singular value for Base Column
    base_value = inp.loc[:,base_column].values[0]
    
    ### Generate DataFrame from the above custom logic to return
    data = [base_value,zeros_lst]
    
    res = pd.DataFrame(data,column_lst).T
       
    return res

实施

from functools import partial
import pandas as pd
import numpy as np


input_list = [
  (11,1)
  ,(11,2)
  ,(11,7)
  ,(45,7)
  ]

sparkDF = sql.createDataFrame(input_list,['ID','CTA'])

schema = StructType([
           StructField('ID', DoubleType(), True),
           StructField('CTA', ArrayType(DoubleType()), True)
 ])

partial_func = partial(generate_ctas_list,base_column='ID',target_column='CTA',column_lst=['ID','CTA'])

sparkDF_agg = sparkDF.groupby('ID').applyInPandas(partial_func,schema)

sparkDF_agg.show(truncate=False)

+----+-----------------------------------+
|ID  |CTA                                |
+----+-----------------------------------+
|11.0|[1.0, 1.0, 0.0, 0.0, 0.0, 0.0, 1.0]|
|45.0|[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 1.0]|
+----+-----------------------------------+

【讨论】：