在 PySpark 中将 StringType 转换为 ArrayType
Posted
技术标签:
【中文标题】在 PySpark 中将 StringType 转换为 ArrayType【英文标题】:Convert StringType to ArrayType in PySpark 【发布时间】:2018-04-05 21:31:05 【问题描述】:我正在尝试在我的数据集上运行 PySpark 中的 FPGrowth 算法。
from pyspark.ml.fpm import FPGrowth
fpGrowth = FPGrowth(itemsCol="name", minSupport=0.5,minConfidence=0.6)
model = fpGrowth.fit(df)
我收到以下错误:
An error occurred while calling o2139.fit.
: java.lang.IllegalArgumentException: requirement failed: The input
column must be ArrayType, but got StringType.
at scala.Predef$.require(Predef.scala:224)
我的 Dataframe df 格式为:
df.show(2)
+---+---------+--------------------+
| id| name| actor|
+---+---------+--------------------+
| 0|['ab,df']| tom|
| 1|['rs,ce']| brad|
+---+---------+--------------------+
only showing top 2 rows
如果我在“名称”列中的数据采用以下形式,则 FP 算法有效:
name
[ab,df]
[rs,ce]
我如何以这种从 StringType 转换为 ArrayType 的形式获得它
我从我的 RDD 中形成了 Dataframe:
rd2=rd.map(lambda x: (x[1], x[0][0] , [x[0][1]]))
rd3 = rd2.map(lambda p:Row(id=int(p[0]),name=str(p[2]),actor=str(p[1])))
df = spark.createDataFrame(rd3)
rd2.take(2):
[(0, 'tom', ['ab,df']), (1, 'brad', ['rs,ce'])]
【问题讨论】:
【参考方案1】:用逗号分隔数据框 name
列中的每一行。 例如
from pyspark.sql.functions import pandas_udf, PandasUDFType
@pandas_udf('list', PandasUDFType.SCALAR)
def split_comma(v):
return v[1:-1].split(',')
df.withColumn('name', split_comma(df.name))
或者更好,不要推迟这个。将名称直接设置为列表。
rd2 = rd.map(lambda x: (x[1], x[0][0], x[0][1].split(',')))
rd3 = rd2.map(lambda p:Row(id=int(p[0]), name=p[2], actor=str(p[1])))
【讨论】:
【参考方案2】:基于your previous question,您似乎错误地构建了rdd2
。
试试这个:
rd2 = rd.map(lambda x: (x[1], x[0][0] , x[0][1].split(",")))
rd3 = rd2.map(lambda p:Row(id=int(p[0]), name=p[2], actor=str(p[1])))
变化是我们在x[0][1]
上调用str.split(",")
,以便它将像'a,b' 这样的字符串转换为列表:['a', 'b']
。
【讨论】:
以上是关于在 PySpark 中将 StringType 转换为 ArrayType的主要内容,如果未能解决你的问题,请参考以下文章
如何在 PySpark 中将字符串转换为字典 (JSON) 的 ArrayType
在 PySpark 中将 URI 查询字符串转换为结构键值数组
每当我尝试在 PySpark 中将字符串转换为 DecimalType 时返回空值
pySpark:java.lang.UnsupportedOperationException:未实现类型:StringType