creating spark data structure from multiline record

by

PySpark since version 1.1 supports Hadoop Input Formats.You can use textinputformat.record.delimiter option to use a custom format delimiter as below

from operator import itemgetter

retrosheet = sc.newAPIHadoopFile(
    '/path/to/retrosheet/file',
    'org.apache.hadoop.mapreduce.lib.input.TextInputFormat',
    'org.apache.hadoop.io.LongWritable',
    'org.apache.hadoop.io.Text',
    conf={'textinputformat.record.delimiter': '\nid,'}
)
(retrosheet
    .filter(itemgetter(1))
    .values()
    .filter(lambda x: x)
    .map(lambda v: (
        v if v.startswith('id') else 'id,{0}'.format(v)).splitlines()))

Since Spark 2.4 you can also read data into DataFrame using text reader

spark.read.option("lineSep", '\nid,').text('/path/to/retrosheet/file')

More Related Contents:

  1. Spark Equivalent of IF Then ELSE
  2. How to add a constant column in a Spark DataFrame?
  3. How to change dataframe column names in pyspark?
  4. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  5. Explode in PySpark
  6. Python worker failed to connect back
  7. How can we JOIN two Spark SQL dataframes using a SQL-esque “LIKE” criterion?
  8. Spark RDD to DataFrame python
  9. Filtering a Pyspark DataFrame with SQL-like IN clause
  10. Spark union of multiple RDDs
  11. Dividing complex rows of dataframe to simple rows in Pyspark
  12. Spark DataFrame TimestampType – how to get Year, Month, Day values from field?
  13. Why is Apache-Spark – Python so slow locally as compared to pandas?
  14. Spark add new column to dataframe with value from previous row
  15. Spark groupByKey alternative
  16. PySpark create new column with mapping from a dict
  17. Tuning parameters for implicit pyspark.ml ALS matrix factorization model through pyspark.ml CrossValidator
  18. How to create a custom Estimator in PySpark
  19. Pyspark: Replacing value in a column by searching a dictionary
  20. Spark iteration time increasing exponentially when using join
  21. Pivot String column on Pyspark Dataframe
  22. Reduce a key-value pair into a key-list pair with Apache Spark
  23. Getting Spark, Python, and MongoDB to work together
  24. How do I set the driver’s python version in spark?
  25. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  26. PySpark in iPython notebook raises Py4JJavaError when using count() and first()
  27. pyspark parse fixed width text file
  28. PySpark: match the values of a DataFrame column against another DataFrame column
  29. When to cache a DataFrame?
  30. environment variables PYSPARK_PYTHON and PYSPARK_DRIVER_PYTHON

Leave a Comment