Чтение файла последовательности в PySpark 2.0

У меня есть файл последовательности, значения которого выглядят как

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

or 

m = sc.parallelize([(1, 2), (3, 4)])
m.map(lambda x,y: str(y)).collect()

Спасибо!


apache-spark pyspark sequencefile
person Max    schedule 09.01.2017    source источник

Ответы (2)


arrow_upward
5
arrow_downward

Фундаментальная проблема с вашим кодом — это функция, которую вы используете. Функция, переданная map, должна принимать один аргумент. Используйте либо:

reader.map(lambda x: x[1])

или просто:

reader.values()

Пока keyClass и valueClass соответствуют данным, это должно быть все, что вам нужно здесь, и не должно быть необходимости в дополнительных преобразованиях типов (это обрабатывается внутри sequenceFile). Напишите на Скале:

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.

Читать на Питоне:

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)

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

person zero323    schedule 09.01.2017

arrow_upward
0
arrow_downward

Для Spark 2.4.x необходимо получить объект sparkContext из SparkSession (объект искры). Который имеет API sequenceFile для чтения файлов последовательности.

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 и создайте кортеж из пары (Key,Value) перед сохранением в виде SequenceFile.

Я надеюсь, что этот ответ поможет вашей цели.

person vijayraj34    schedule 07.12.2020