我有一个序列文件,其值如下
(string_value, json_value)
我不关心字符串值.
在Scala我可以通过阅读文件
val reader = sc.sequenceFile[String, String]("/path...")
val data = reader.map{case (x, y) => (y.toString)}
val jsondata = spark.read.json(data)
我很难将其转换为PySpark.我试过用
reader= sc.sequenceFile("/path","org.apache.hadoop.io.Text", "org.apache.hadoop.io.Text")
data = reader.map(lambda x,y: str(y))
jsondata = spark.read.json(data)
这些错误很神秘,但如果有帮助我可以提供.我的问题是,在pySpark2中读取这些序列文件的正确语法是什么？
我想我没有正确地将数组元素转换为字符串.如果我做一些简单的事情,我会得到类似的错误
m = sc.parallelize([(1, 2), (3, 4)])
m.map(lambda x,y: y.toString).collect()
要么
m = sc.parallelize([(1, 2), (3, 4)])
m.map(lambda x,y: str(y)).collect()
谢谢!
您的代码的基本问题是您使用的功能.传递给的函数map应该采用单个参数.使用:
reader.map(lambda x: x[1])
要不就:
reader.values()
只要keyClass和valueClass匹配数据,这应该是你需要的所有内容,并且不需要额外的类型转换(这由内部处理sequenceFile).用Scala写:
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.1.0
/_/
Using Scala version 2.11.8 (OpenJDK 64-Bit Server VM, Java 1.8.0_111)
Type in expressions to have them evaluated.
Type :help for more information.
scala> :paste
// Entering paste mode (ctrl-D to finish)
sc
.parallelize(Seq(
("foo", """{"foo": 1}"""), ("bar", """{"bar": 2}""")))
.saveAsSequenceFile("example")
// Exiting paste mode, now interpreting.
读入Python:
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/__ / .__/\_,_/_/ /_/\_\ version 2.1.0
/_/
Using Python version 3.5.1 (default, Dec 7 2015 11:16:01)
SparkSession available as 'spark'.
In [1]: Text = "org.apache.hadoop.io.Text"
In [2]: (sc
...: .sequenceFile("example", Text, Text)
...: .values()
...: .first())
Out[2]: '{"bar": 2}'
注意:
旧版Python版本支持元组参数解包:
reader.map(lambda (_, v): v)
不要将它用于应向前兼容的代码.
京公网安备 11010802040832号 | 京ICP备19059560号-6 
