Parsing multiline records in Scala

by

By default Spark creates a single element per line. It means that in your case every record is spread over multiple elements which, as stated by Daniel Darabos in the comments, can be processed by different workers.

Since it looks like your data is relatively regular and separated by an empty line you should be able to use newAPIHadoopFile with custom delimiter:

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

val path: String = ???

val conf = new org.apache.hadoop.mapreduce.Job().getConfiguration
conf.set("textinputformat.record.delimiter", "\n\n")

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

val usgPairRDD: RDD[(String, Seq[String])] = usgRDD.map(_.split("\n") match {
  case Array(x, xs @ _*) => (x, xs)
})

In Spark 2.4 or later data loading part can be also achieved using Dataset API:

val ds: Dataset[String] = spark.read.option("lineSep", "\n\n").text(path)

More Related Contents:

  1. How does HashPartitioner work?
  2. How to convert rdd object to dataframe in spark
  3. (Why) do we need to call cache or persist on a RDD
  4. Case class equality in Apache Spark
  5. Spark performance for Scala vs Python
  6. Stackoverflow due to long RDD Lineage
  7. How to find spark RDD/Dataframe size?
  8. Modify collection inside a Spark RDD foreach
  9. Explanation of fold method of spark RDD
  10. Why does partition parameter of SparkContext.textFile not take effect?
  11. Why does Spark RDD partition has 2GB limit for HDFS?
  12. Reading in multiple files compressed in tar.gz archive into Spark [duplicate]
  13. How to sort an RDD in Scala Spark?
  14. How to transpose an RDD in Spark
  15. Apache Spark: map vs mapPartitions?
  16. How to define schema for custom type in Spark SQL?
  17. NullPointerException in Scala Spark, appears to be caused be collection type?
  18. What does setMaster `local[*]` mean in spark?
  19. How can I pass extra parameters to UDFs in Spark SQL?
  20. How to split a dataframe into dataframes with same column values?
  21. How to use regex to include/exclude some input files in sc.textFile?
  22. How to define a custom aggregation function to sum a column of Vectors?
  23. How to write spark streaming DF to Kafka topic
  24. Better way to convert a string field into timestamp in Spark
  25. Encode an ADT / sealed trait hierarchy into Spark DataSet column
  26. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  27. Replace null values in Spark DataFrame
  28. Spark RDD default number of partitions
  29. Calculate Cosine Similarity Spark Dataframe
  30. Scala-Spark Dynamically call groupby and agg with parameter values

Leave a Comment