Xml processing in Spark

by

Yes it possible but details will differ depending on an approach you take.

  • If files are small, as you’ve mentioned, the simplest solution is to load your data using SparkContext.wholeTextFiles. It loads data as RDD[(String, String)] where the the first element is path and the second file content. Then you parse each file individually like in a local mode.
  • For larger files you can use Hadoop input formats.
    • If structure is simple you can split records using textinputformat.record.delimiter. You can find a simple example here. Input is not a XML but it you should give you and idea how to proceed
    • Otherwise Mahout provides XmlInputFormat

  • Finally it is possible to read file using SparkContext.textFile and adjust later for record spanning between partitions. Conceptually it means something similar to creating sliding window or partitioning records into groups of fixed size:

    • use mapPartitionsWithIndex partitions to identify records broken between partitions, collect broken records
    • use second mapPartitionsWithIndex to repair broken records

Edit:

There is also relatively new spark-xml package which allows you to extract specific records by tag:

val df = sqlContext.read
  .format("com.databricks.spark.xml")
   .option("rowTag", "foo")
   .load("bar.xml")

More Related Contents:

  1. Find maximum row per group in Spark DataFrame
  2. Multiple Aggregate operations on the same column of a spark dataframe
  3. Partitioning in spark while reading from RDBMS via JDBC
  4. How to read multiple text files into a single RDD?
  5. How to access element of a VectorUDT column in a Spark DataFrame?
  6. What is the difference between map and flatMap and a good use case for each?
  7. Python Spark Cumulative Sum by Group Using DataFrame
  8. Link Spark with iPython Notebook
  9. Default Partitioning Scheme in Spark
  10. What is the maximum size for a broadcast object in Spark?
  11. Any performance issues forcing eager evaluation using count in spark?
  12. How does createOrReplaceTempView work in Spark?
  13. Debugging “Managed memory leak detected” in Spark 1.6.0
  14. Spark: how to get the number of written rows?
  15. PySpark – get row number for each row in a group
  16. Why are Spark Parquet files for an aggregate larger than the original?
  17. Save ML model for future usage
  18. Determining optimal number of Spark partitions based on workers, cores and DataFrame size
  19. Apache spark dealing with case statements
  20. pyspark: count distinct over a window
  21. Mind blown: RDD.zip() method
  22. How spark read a large file (petabyte) when file can not be fit in spark’s main memory
  23. Serialize a custom transformer using python to be used within a Pyspark ML pipeline
  24. Not able to cat dbfs file in databricks community edition cluster. FileNotFoundError: [Errno 2] No such file or directory:
  25. What do the numbers on the progress bar mean in spark-shell?
  26. pyspark: rolling average using timeseries data
  27. How to improve broadcast Join speed with between condition in Spark
  28. Spark ALS predictAll returns empty
  29. Spark SQL referencing attributes of UDT
  30. Why is the fold action necessary in Spark?

Leave a Comment