如何使用 pyspark SQL 将字符串传输到字典答案

【问题标题】：How to transfer the string to a dict with pysparkSQL如何使用 pyspark SQL 将字符串传输到字典
【发布时间】：2018-09-06 11:47:12
【问题描述】：

在 pysparkSQL 中，我有一个名为 bmd2 的 DataFrame，如下所示：

DataFrame[genres: string, id: int, tagline: string, title: string, vote_average: double, vote_count: int]

而bmd2['genres'] 的数据是这样的：

bmd2.select('genres').show():

+--------------------+
|              genres|
+--------------------+
|[{'id': 16, 'name...|
|[{'id': 12, 'name...|
|[{'id': 10749, 'n...|
|[{'id': 35, 'name...|
|[{'id': 35, 'name...|
|[{'id': 28, 'name...|
|[{'id': 35, 'name...|
|[{'id': 28, 'name...|
|[{'id': 28, 'name...|
|[{'id': 12, 'name...|
|[{'id': 35, 'name...|
|[{'id': 35, 'name...|
|[{'id': 10751, 'n...|
|[{'id': 36, 'name...|
|[{'id': 28, 'name...|
|[{'id': 18, 'name...|
|[{'id': 18, 'name...|
|[{'id': 80, 'name...|
|[{'id': 80, 'name...|
|[{'id': 28, 'name...|
+--------------------+
only showing top 20 rows

“流派”列中的数据类型是字符串，但可以在 python 中使用“eval 函数”将它们传输到字典列表。那么我应该如何在这里应用 eval() 来将这里的字符串传输到每一行中？我尝试了很多方法：

bmd2.select('genres'.astype('list')):AttributeError: 'str' object 没有属性“astype”

bmd2.select(eval('genres')):NameError: name 'genres' is not defined

bmd2.withColumn('genres',eval('genres')):NameError: name 'genres' 未定义

【问题讨论】：

你想在哪里存储字典列表？
仍在'流派'或新列中，都可以
我们可以看到没有截断的数据帧吗？如果您也提供示例输出会很有帮助

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

【解决方案1】：

我使用 UDF 解决了我的问题，这是用户定义的函数。

首先，导入它：

from pyspark.sql.functions import udf

然后，定义你的 UDF，就像一个匿名函数：

getdirector = udf(lambda x:[i['name'] for i in x if i['job'] == 'Director'],StringType())

你应该在这里指定返回值的类型，这样你就会得到一个你期望的类型的返回值。然后你可以像其他函数一样在你的代码中调用这个 UDF。

cres2 = cres1.select('id',getcharacter('cast').alias('cast'),getdirector('crew').alias('crew'))

在这个问题中，我可以修改 UDF 以获得我需要的任何类型。

【讨论】：

【解决方案2】：

我写这个作为答案，因为我找不到评论选项。我建议您从 pyspark.sql.functions 中查看 from_json。例如，这就是你将如何使用它：

# given a row that looks like:

+----------genres-------------+
| [{ id:1, name:"hiphop"}]    |
+-----------------------------+

# define a schema
schema = ArrayType(StructType().add("id", IntegerType())\
                              .add("name", StringType()))

# transform
new_df = df.select(from_json("genres", schema).alias("genres_dict"))

# display
new_df.printSchema()
new_df.show()

还有另一种方法可以使用名为 regexp_extract 的函数来实现此目的。但以上是我个人的喜好。此外，如果您想切换回原始字符串，您可以使用 to_json 函数。希望这会有所帮助。

【讨论】：