主页 / coding / 问题 / 78572920
Accepted
Mikhail Genkin
Asked: 2024-06-04 08:19:03 +0800 CST

在 pyspark 中将许多二进制列折叠成单个列

  • 772

在我的 pyspark 作业中，我有一个巨大的数据框架，其中有超过 6,000 列，格式如下：

id_        a1   a2     a3   a4     a5   .... a6250
u1827s    True False  True False False .... False
...

其中大多数列a1,a2,a3,...,a6250 都是二进制类型。我需要按所有这些列对这些数据进行分组，并聚合每个组合的不同 id 的数量，例如

df = df.groupby(list_of_cols).agg(F.countDistinct("id_"))

在哪里list_of_cols = [a1,a2,...,a6250]。运行此 pyspark 作业时，出现java.lang.StackOverflowError错误。我知道我可以增加堆栈大小（按照https://rangereddy.github.io/SparkStackOverflow/），但是，我更喜欢一个更优雅的解决方案，也可以实现更方便的输出。

在分组之前我有两个想法：

  1. 将 a1,a2,...,a6250 列的组合编码为单个二进制列，例如具有 6250 位的二进制数，其中位置上的位k将为该列编码 True 或 False 值a_k，例如在上面的示例中的值将是10100...0（a1 为真，a2 为假，a3 为真，a4 为假，a5 为假，...a6250 为假）。

  2. 将这些值收集到一个二进制数组中，例如有 1 列，如数组（True，False，True，False，False，....，False）。

哪种方法更好 - 增加堆栈大小并处理 6000 多个列、使用单个二进制列还是二进制值数组？

pyspark

1 个回答

  1. Best Answer

    我相信第一个解决方案对于找到所有现有组合的不同计数会更实用。这是我尝试根据您的数据架构重现此解决方案的一些代码

    from pyspark.sql import Row
import pyspark.sql.functions as F

data = [
    ("A", True, True, True),  # Counted
    ("A", True, True, True),  # Not Counted
    ("A", False, False, False),  # Counted
    ("B", False, False, False),  # Counted
    ("B", False, False, False),  # Not Counted
    ("B", True, True, True),  # Counted
]

columns = ["id", "a0", "a1", "a2"]
df = spark.createDataFrame(data, columns)

# To dynamically access the columns list on each executor
dynamic_column_list = spark.sparkContext.broadcast(df.columns)


def extract_pattern(row):
    """
    Extract the information from the row according to the following values and append to pattern.

    True => "1"
    False => "0"
    """
    pattern = ""
    for column in dynamic_column_list.value:
        bit = ""
        if row[column] is True:
            bit = "1"
        elif row[column] is False:
            bit = "0"
        pattern += bit
    return Row(id=row["id"], pattern=pattern)


extracted_pattern = df.rdd.map(lambda r: extract_pattern(r)).toDF()
counted_by_pattern = extracted_pattern.groupBy("pattern").agg(F.countDistinct("id"))
counted_by_pattern.show()

    我尝试使用相同的函数来获取与每个模式关联的 ID 的不同计数

    counted_by_pattern = extracted_pattern.groupBy("pattern").agg(F.countDistinct("id"))

    我还没有尝试复制 6K+ 列，但请在您查看并尝试此代码后分享结果。

    • 1

相关问题