在 PySpark 2.0 中读取序列文件

Question

我有一个序列文件，其值类似于

(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()

谢谢！

Answer 1

您的代码的根本问题在于您使用的函数。传递给 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)

不要将它用于应该向前兼容的代码。

Answer 2

对于 Spark 2.4.x，您必须从 SparkSession（spark 对象）中获取 sparkContext 对象。其中有 sequenceFile API 来读取序列文件。

spark.
sparkContext.
sequenceFile('/user/sequencesample').
toDF().show()

以上一个很有魅力。

写入（parquet到sequenceFile）：

spark.
read.
format('parquet').
load('/user/parquet_sample').
select('id',F.concat_ws('|','id','name')).
rdd.map(lambda rec:(rec[0],rec[1])).
saveAsSequenceFile('/user/output')

首先将 DF 转换为 RDD，并在保存为 SequenceFile 之前创建一个 (Key,Value) 对的元组。

希望这个回答对您有所帮助。

在 PySpark 2.0 中读取序列文件

Reading Sequence File in PySpark 2.0

sequencefile

apache-spark

pyspark