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

Ответ 1

В оболочке Spark я извлек данные в соответствии с Установка textinputformat.record.delimiter в искровом режиме:

$ spark-shell
...
scala> import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.LongWritable

scala> import org.apache.hadoop.io.Text
import org.apache.hadoop.io.Text

scala> import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.conf.Configuration

scala> import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat

scala> val conf = new Configuration
conf: org.apache.hadoop.conf.Configuration = Configuration: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, yarn-site.xml

scala> conf.set("textinputformat.record.delimiter", "\u0001")

scala> val data = sc.newAPIHadoopFile("mydata.txt", classOf[TextInputFormat], classOf[LongWritable], classOf[Text], conf).map(_._2.toString)
data: org.apache.spark.rdd.RDD[(org.apache.hadoop.io.LongWritable, org.apache.hadoop.io.Text)] = NewHadoopRDD[0] at newAPIHadoopFile at <console>:19

sc.newAPIHadoopFile("mydata.txt", ...) - это RDD[(LongWritable, Text)], где первая часть элементов является начальным символьным индексом, а вторая часть является фактическим текстом, разделенным символом "\u0001".

Ответ 2

Вы можете использовать textinputformat.record.delimiter для установки разделителя для TextInputFormat, например,

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat

val conf = new Configuration(sc.hadoopConfiguration)
conf.set("textinputformat.record.delimiter", "X")
val input = sc.newAPIHadoopFile("file_path", classOf[TextInputFormat], classOf[LongWritable], classOf[Text], conf)
val lines = input.map { case (_, text) => text.toString}
println(lines.collect)

Например, мой ввод - это файл, содержащий одну строку aXbXcXd. Вышеприведенный код выводит

Array(a, b, c, d)

Ответ 3

В python это может быть достигнуто с помощью:

rdd = sc.newAPIHadoopFile(YOUR_FILE, "org.apache.hadoop.mapreduce.lib.input.TextInputFormat",
            "org.apache.hadoop.io.LongWritable", "org.apache.hadoop.io.Text",
            conf={"textinputformat.record.delimiter": YOUR_DELIMITER}).map(lambda l:l[1])

Ответ 4

Вот готовая к использованию версия Chad и @zsxwing для пользователей Scala, которые могут быть использованы следующим образом:

sc.textFile("some/path.txt", "\u0001")

Следующий фрагмент создает дополнительный метод textFile неявно прикрепленный к SparkContext с использованием implicit class (для репликации SparkContext textFile методу SparkContext умолчанию):

package com.whatever

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat

object Spark {

  implicit class ContextExtensions(val sc: SparkContext) extends AnyVal {

    def textFile(
        path: String,
        delimiter: String,
        maxRecordLength: String = "1000000"
    ): RDD[String] = {

      val conf = new Configuration(sc.hadoopConfiguration)

      // This configuration sets the record delimiter:
      conf.set("textinputformat.record.delimiter", delimiter)
      // and this one limits the size of one record:
      conf.set("mapreduce.input.linerecordreader.line.maxlength", maxRecordLength)

      sc.newAPIHadoopFile(
          path,
          classOf[TextInputFormat], classOf[LongWritable], classOf[Text],
          conf
        )
        .map { case (_, text) => text.toString }
    }
  }
}

которые могут быть использованы следующим образом:

import com.whatever.Spark.ContextExtensions
sc.textFile("some/path.txt", "\u0001")

Обратите внимание на дополнительную настройку mapreduce.input.linerecordreader.line.maxlength которая ограничивает максимальный размер записи. Это пригодится при чтении из поврежденного файла, для которого запись может быть слишком длинной, чтобы вписаться в память (больше шансов, что это произойдет при игре с разделителем записей).

С этой настройкой при чтении поврежденного файла исключается исключение (java.io.IOException - таким образом, захватывающее), а не получение беспорядочной памяти, которая остановит SparkContext.