在Pyspark的FPGrowth中将RDD转换为数据帧

Question

在从< code>RDD制作< code >数据帧时，我遇到了一个错误。

from pyspark.ml.fpm import FPGrowth

sogou = sc.textFile("SogouQ.sample.utf8", use_unicode = False)

def parse(line):
    value = [ x for x in line.split(",") if x]
    return list(set(value))

rdd = sogou.map(parse)
df = sogou.toDF('items')

我收到以下错误：

py spark . SQL . utils . parse exception:u " \ nmis matched input ' '应为{'SELECT '，' FROM '，' ADD '，' AS '，' ALL '，' DISTINCT '，' WHERE '，' GROUP '，' BY '，' GROUPING '，' SETS '，' CUBE '，' ROLLUP '，' ORDER '，' HAVING '，' LIMIT '，' AT '，' OR '，' AND '，' IN '，NOT，' NO '，' EXISTS '，' BETWEEN '，' LIKE '，RLIKE，' IS '，' NULL '，' TRUE '，' FALSE '，' NULL '，' ASC '，' FOR '，' INTERVAL FORMAT '，' LOGICAL '，' CODEGEN '，' COST '，' CAST '，' SHOW '，' TABLES '，' COLUMNS '，' DROP '，' UNION '，' EXCEPT '，' MINUS '，' INTERSECT '，' TO '，' TABLESAMPLE '，' STRATIFY '，' ALTER '，' RENAME '，' ARRAY '，' DATA '，' START '，' TRANSACTION '，' COMMIT '，' ROLLBACK '，' MACRO '，' IGNORE '，' BOTH '，' BOTH '，' LEADING '，' TRAILING '，' IF '，' POSITION '，' DIV '，' percenter 目录'，'位置'，'交换'，'存档'，'取消存档'，'文件格式'，'触摸'，'压缩'，'连接'，'更改'，'级联'，'限制'，'聚集'，'排序'，'清除'，'输入格式'，'数据库，数据库，' DFS '，'截断'，'分析'，'计算'，'列表'，'统计'，'分区'，'外部'，'定义'，'撤销'，'授予'，'锁定'，'解锁'，' MSCK '，'修复'，'恢复'，'导出'，'导入'，'加载'，'角色'，'角色'，'压缩'，'委托人'，'事务'，'索引'，'索引'，'

文本包含中文。有关系吗？文字是这样的：

360,安全卫士，
123，123，范冰冰，

当我使用< code > py spark . ml lib . FP growth 时，< code>rdd工作正常。怎么转换成dataframe？

Answer 1

这里有两个不同的问题：

> < li>

toDF调用。< code>RDD.toDF具有以下签名:

Signature: rdd.toDF(schema=None, sampleRatio=None)

架构应为的位置

参数模式：pyspark.sql.types.结构类型或列名列表

所以在你的情况下，它应该是：

sogou.toDF(["items"])

解析方法：

createDataFrame由df调用的方法需要一个RDD[tuple]或等价物，它可以映射到结构，除非提供了架构。如果您只想使用名称，它应该返回一个tuple

def parse(line):
    value = [ x for x in line.split(",") if x]
    return list(set(value)),  

组合:

>>> def parse(line):
...     value = [ x for x in line.split(",") if x]
...     return list(set(value)),  
... 
... 
>>> rdd = sc.parallelize(["360,安全卫士,", "123，123，范冰冰，"])
>>> rdd.map(parse).toDF(["items"]).show()
+--------------+
|         items|
+--------------+
|   [安全卫士, 360]|
|[123，123，范冰冰，]|
+--------------+

另一种选择（保留当前的解析实现）是

>>> from pyspark.sql.types import ArrayType, StringType
>>> def parse(line):
...     value = [ x for x in line.split(",") if x]
...     return list(set(value))
    >>> rdd.map(parse).toDF(ArrayType(StringType())).toDF("items").show()
+--------------+     
|         items|
+--------------+
|   [安全卫士, 360]|
|[123，123，范冰冰，]|
+--------------+