简化一个GO中的多个（30列）列复杂的Pyspark聚合

Question

我有一个示例火花DF如下：

df = ([[1, 'a', 'b' , 'c'],
      [1, 'b', 'c' , 'b'],
      [1, 'b', 'a' , 'b'],
      [2, 'c', 'a' , 'a'],
      [3, 'b', 'b' , 'a']]).toDF(['id', 'field1', 'field2', 'field3'])

下一步需要的是提供多个聚合，以显示每个字段的A，B，C值的摘要。我有一个工作，但乏味的过程如下：

agg_table = (
  df
        .groupBy('id') 
        .agg(
          # field1
             sum(when(col('field1') == 'a',1).otherwise(0)).alias('field1_a_count')
             ,sum(when(col('field1') == 'b',1).otherwise(0)).alias('field1_b_count')
             ,sum(when(col('field1') == 'c',1).otherwise(0)).alias('field1_c_count')
          # field2
             ,sum(when(col('field2') == 'a',1).otherwise(0)).alias('field2_a_count')
             ,sum(when(col('field2') == 'b',1).otherwise(0)).alias('field2_b_count')
             ,sum(when(col('field2') == 'c',1).otherwise(0)).alias('field2_c_count')
          # field3
             ,sum(when(col('field3') == 'a',1).otherwise(0)).alias('field3_a_count')
             ,sum(when(col('field3') == 'b',1).otherwise(0)).alias('field3_b_count')
             ,sum(when(col('field3') == 'c',1).otherwise(0)).alias('field3_c_count')
         ))

我希望得到的是：

agg_table = (['id':'1','2','3'],
             ['field1_a_count':1,0,0],
             ['field1_b_count':2,0,1],
             ['field1_c_count':0, 1, 0],
             ['field2_a_count':1,1,0],
             ['field2_b_count':1,0,1],
             ['field2_c_count':1,0,0],
             ['field3_a_count':0,1,1],
             ['field3_b_count':2,0,0],
             ['field3_c_count':1,0,0])

如果我只有3个字段，那很好，但是我有30个字段，具有不同的/自定义名称。也许有人可以帮助我完成编码每个字段总和的重复任务。我试图探讨以下建议：

我可以使它起作用

AnalysisException: cannot resolve '`value`' given input columns: ['field1','field2','field3']

如果我只能拉一个列和一个值， 尝试使用的是：

validated_cols = ['field1','field2','field3']

df.select(validated_cols).groupBy('id').agg(collect_list($'field1_a_count',$'field1_b_count',$'field1_c_count', ...
 
 $'field30_c_count')).show()

Output: SyntaxError: invalid syntax

我也尝试使用枢轴，但是从搜索到目前为止，它说它仅适用于一列。我尝试了多个列：

df.withColumn("p", concat($"p1", $"p2"))
  .groupBy("a", "b")
  .pivot("p")
  .agg(...)

我仍然会遇到语法错误。

我尝试过的另一个链接： https：// blog/2019/02/05/complect-aggregations in-pyspark/

我还尝试了exprs方法： exprs1 = {x：x in x in df.columns in x！='id in df.columns中'}

将不胜感激。谢谢

Answer 1

让我分两个步骤回答您的问题。首先，您想知道是否有可能避免硬编码所有聚合，以试图计算所有聚合。这是。我会这样这样做：

from pyspark.sql import functions as f

# let's assume that this is known, but we could compute it as well
values = ['a', 'b', 'c']
# All the columns except the id
cols = [ c for c in df.columns if c != 'id' ]

def count_values(column, value):
    return f.sum(f.when(f.col(column) == value, 1).otherwise(0))\
            .alias(f"{column}_{value}_count")

# And this gives you the result of your hard coded aggregations:
df\
    .groupBy('id')\
    .agg(*[count_values(c, value) for c in cols for value in values])\
    .show()

但这不是您期望的正确吗？您正在尝试在 ID 列上计算某种枢轴。为此，我不会使用先前的结果，而只是使用不同的数据来处理数据。首先，我将通过 id （将其更名为 x ）替换 ID ），从表单 {column_name} _ { value} _count ，我会爆炸该数组。从那里开始，我们只需要在以前的 id 列上计算一个简单的枢轴，重命名为 x ，由爆炸数组中包含的值分组。

df\
    .select(f.col('id').alias('x'), f.explode(
         f.array(
             [f.concat_ws('_', f.lit(c), f.col(c), f.lit('count')).alias(c)
                 for c in cols]
         )
    ).alias('id'))\
    .groupBy('id')\
    .pivot('x')\
    .count()\
    .na.fill(0)\
    .orderBy('id')\
    .show()

哪个产生：

+--------------+---+---+---+
|            id|  1|  2|  3|
+--------------+---+---+---+
|field1_a_count|  1|  0|  0|
|field1_b_count|  2|  0|  1|
|field1_c_count|  0|  1|  0|
|field2_a_count|  1|  1|  0|
|field2_b_count|  1|  0|  1|
|field2_c_count|  1|  0|  0|
|field3_a_count|  0|  1|  1|
|field3_b_count|  2|  0|  0|
|field3_c_count|  1|  0|  0|
+--------------+---+---+---+

Answer 2

根据评论中的讨论的更新

，我认为这个问题是 xy问题< /a>。手头的任务是在数据工程和ETL开发的世界中经常看到的东西：如何分区然后量化好和坏记录。

如果准备数据将数据加载到数据仓库 / Hadoop生态系统的情况下，通常的模式是获取原始输入并将其加载到数据范围内，然后应用转换＆amp;将数据划分为“好，坏和丑陋”的验证：

• 第一个，希望最大的分区包含成功转换和通过验证的记录。这些将继续在耐用的存储中持续存在，并经过认证，可用于Anayltics。
• 第二个分区包含成功转换但在质量检查期间失败的记录。质量检查规则应包括对非法空的检查，字符串模式匹配（例如电话号码格式）等...
• 第三个分区是用于在过程中早期拒绝的记录，因为它们在转换步骤中失败。示例包括包含非数值值的字段，这些字段被铸造为数字类型，超过最大长度的文本字段或包含数据库不支持的控制字符的字符串。

目标不应是在每一列和每一行中为这三个分类中的每一个生成计数。试图做到这是适得其反的。为什么？因为当转换步骤或QA检查失败以获取给定记录时，应立即拒绝整个记录并将其发送到单独的输出流以稍后进行分析。数据集中的每一行应仅视为：一个记录。单个字段不可能失败，并且仍然拥有完整的记录通行证，这使该粒度不必要。您知道100行在“地址”字段上通过，您会采取什么行动？对于有效的记录，重要的是每列传递的总数。否则，这将不是有效的记录。

话虽如此，请记住，目标是构建可用且清洁的数据集；分析被拒绝的记录是次要任务，可以离线完成。

通常的做法是将字段添加到被拒绝的数据以指示哪个列导致故障。这使得对任何错误的数据进行故障排除变得容易，因此即使对于不良记录，也无需在所有列中生成计数。取而代之的是，只需在主要工作完成后查看被拒绝的数据，然后解决问题。继续迭代地进行此操作，直到被拒绝的记录的数量低于您认为合理的任何门槛，然后继续监视它。

---

旧答案

这是数据中设计缺陷的迹象。无论“ field1”，“ field2”等...列实际上代表了什么，看来它们都是相关的，从某种意义上说，值量化了某些属性（也许每个属性都是特定商品ID或数字的计数拥有一定财产的人...）。问题在于，这些字段在事实表 ¹ 上添加为单个列，然后需要汇总，从而导致您面临的情况。

一个更好的设计是将那些“ field1”，“ field2”等折叠成一个单个代码字段，该字段可以用作 five在执行聚合时用作组。如果现有的列有许多其他列，并且进行此更改会以可能导致其他问题的方式改变谷物，则可能需要考虑创建一个单独的表来执行此操作。

---

^{1：通常是一个带有一堆具有相同名称和目的的列表的表，这通常是一个大的危险信号。我什至看到了某人创建了带有“备用”列的表格的情况，以便他们以后想添加更多属性。不好。}