在dataframe中创建字典类型列

Question

请考虑以下数据：

------------+--------------------+
|id|          values
+------------+--------------------+
|          39|a,a,b,b,c,c,c,c,d
|         520|a,b,c
|         832|a,a

我想将其转换为以下DataFrame：

------------+--------------------+
|id|          values
+------------+--------------------+
|          39|{"a":2, "b": 2,"c": 4,"d": 1}
|         520|{"a": 1,"b": 1,"c": 1}
|         832|{"a": 2}

我尝试了两种方法：

1. 将dataframe转换为rdd。然后，我将值列映射到一个频率计数器函数。但是，在将rdd转换回dataframe时会出现错误。
2. 使用udf在本质上做与上面相同的事情。

我想要一个字典列的原因是在我的python应用程序中将它作为json加载。

Answer 1

您可以使用返回MapType列的udf来完成此操作。

from pyspark.sql.types import MapType, StringType, IntegerType
from collections import Counter

my_udf = udf(lambda s: dict(Counter(s.split(','))), MapType(StringType(), IntegerType()))
df = df.withColumn('values', my_udf('values'))
df.collect()

[Row(id=39, values={u'a': 2, u'c': 4, u'b': 2, u'd': 1}),
 Row(id=520, values={u'a': 1, u'c': 1, u'b': 1}),
 Row(id=832, values={u'a': 2})]

Answer 2

我不能完全得到你所需要的输出，但我真的很接近。这就是我能得到的：

from pyspark.sql.functions import explode, split
counts = (df.select("id", explode(split("values", ",")).alias("value")).groupby("id", "value").count())
counts.show()

输出：

+---+-----+-----+
| id|value|count|
+---+-----+-----+
|520|    a|    1|
|520|    b|    1|
|520|    c|    1|
| 39|    a|    2|
| 39|    b|    2|
| 39|    c|    4|
| 39|    d|    1|
|832|    a|    2|
+---+-----+-----+

可能有人可以添加它所需的东西，以获得所需的输出。希望能帮上忙。

Answer 3

我最终使用了这个；如果您觉得有更好的方法，请告诉我。

def split_test(str_in):
  a = str_in.split(',')
  b = {}
  for i in a:
    if i not in b:
      b[i] = 1
    else:
      b[i] += 1

  return str(b)

udf_value_count = udf(split_test, StringType() )

value_count_df = value_df.withColumn('value_count', udf_value_count(value_df.values)).drop('values')