pyspark 从逗号分隔值列表中创建多行

Question

在pyspark sqlcontext sql中，已经编写了获取文本然后重新格式化的代码 但是这样的事情是个问题

在数据框中有这样的东西 代码在哪里

hash_tags_fun = udf(lambda t: re.findall('(#[^#]\w{3,})', t))

hash_tags_in_tweets_df.registerTempTable("hash_tags_table")
hash_tags_result = sqlContext.sql("SELECT text FROM hash_tags_table")
hash_tags_list = hash_tags_result.withColumn('text', hash_tags_fun('text'))
hash_tags_list.show(3)
+-------------------+
|               text|
+-------------------+
|  [#shutUpAndDANCE]|
|  [#SHINee, #AMBER]|
|[#JR50, #flipagram]|
+-------------------+

我需要类似的东西

+-------------------+
|               text|
+-------------------+
|    #shutUpAndDANCE|
|            #SHINee|
|             #AMBER|
|              #JR50|
|         #flipagram|
+-------------------+

hash_tags_list.withColumn("text", explode("text")) has given an error saying

AnalysisException: u"cannot resolve 'explode(text)' due to data type 不匹配：函数 explode 的输入应该是数组或映射类型，而不是 细绳;; \n'项目 [explode(text#24) AS text#68]\n+- 分析障碍\n
+- 项目 [(text#9) AS text#24]\n
+- 项目 [text#9]\n
+- 子查询别名 hash_tags_table\n
+- 项目 [text#9]\n
+- 过滤文本#9 LIKE %#%\n
+- 子查询别名 twt\n
+- SubqueryAlias 推文\n
+- 关系[country#6,id#7,place#8,text#9,user#10] json\n"

Answer 1

扩展至my comment：

您的列看起来像一个数组，但它实际上是一个字符串 - 这就是您对 explode() 的调用不起作用的原因。您必须先将列转换为数组。

这将涉及删除前导和尾随方括号并拆分逗号字符。

先去掉前后括号，可以使用pyspark.sql.functions.regexp_replace()：

from pyspark.sql.functions import regexp_replace, split
df = hash_tags_list.select(regexp_replace("text", r"(^\[)|(\]$)", "").alias("text"))
df.show()
#+-----------------+
#|             text|
#+-----------------+
#|  #shutUpAndDANCE|
#|  #SHINee, #AMBER|
#|#JR50, #flipagram|
#+-----------------+

现在以逗号分隔，后跟一个空格：

df = df.select(split("text", ", ").alias("text"))
df.show()
#+-------------------+
#|               text|
#+-------------------+
#|  [#shutUpAndDANCE]|
#|  [#SHINee, #AMBER]|
#|[#JR50, #flipagram]|
#+-------------------+

你会注意到这个打印和你开始的一模一样，但是当我们检查架构时，我们发现这些实际上是字符串数组：

df.printSchema()
#root
# |-- text: array (nullable = true)
# |    |-- element: string (containsNull = true)

将此与您原始 DataFrame 的架构进行比较：

hash_tags_list.printSchema()
#root
# |-- text: string (nullable = true)

将数据作为数组，现在可以调用explode()：

from pyspark.sql.functions import explode
df = df.select(explode("text").alias("hashtags"))
df.show()
#+---------------+
#|       hashtags|
#+---------------+
#|#shutUpAndDANCE|
#|        #SHINee|
#|         #AMBER|
#|          #JR50|
#|     #flipagram|
#+---------------+