Последовательность чтения файлов в 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) 

Не используйте его для кода, который должен быть совместимым с переходом.