Spark-Obtaining file name in RDDs

by

Since Spark 1.6 you can combine text data source and input_file_name function as follows:

Scala:

import org.apache.spark.sql.functions.input_file_name

val inputPath: String = ???

spark.read.text(inputPath)
  .select(input_file_name, $"value")
  .as[(String, String)] // Optionally convert to Dataset
  .rdd // or RDD

Python:

(Versions before 2.x are buggy and may not preserve names when converted to RDD):

from pyspark.sql.functions import input_file_name

(spark.read.text(input_path)
    .select(input_file_name(), "value"))
    .rdd)

This can be used with other input formats as well.

More Related Contents:

  1. How to melt Spark DataFrame?
  2. How to stop INFO messages displaying on spark console?
  3. DataFrame join optimization – Broadcast Hash Join
  4. Overwrite specific partitions in spark dataframe write method
  5. Convert date from String to Date format in Dataframes
  6. What is the meaning of partitionColumn, lowerBound, upperBound, numPartitions parameters?
  7. Spark Transformation – Why is it lazy and what is the advantage?
  8. What is the difference between spark checkpoint and persist to a disk
  9. Spark: Reading files using different delimiter than new line
  10. Spark MLlib LDA, how to infer the topics distribution of a new unseen document?
  11. Adding a group count column to a PySpark dataframe
  12. Why does format(“kafka”) fail with “Failed to find data source: kafka.” (even with uber-jar)?
  13. How to loop through each row of dataFrame in pyspark
  14. NoClassDefFoundError com.apache.hadoop.fs.FSDataInputStream when execute spark-shell
  15. How to group by common element in array?
  16. Spark: Best practice for retrieving big data from RDD to local machine
  17. Understanding Spark serialization
  18. Pyspark: Pass multiple columns in UDF
  19. How do I add an persistent column of row ids to Spark DataFrame?
  20. How does Spark aggregate function – aggregateByKey work?
  21. Rename more than one column using withColumnRenamed
  22. Convert null values to empty array in Spark DataFrame
  23. DataFrame partitionBy to a single Parquet file (per partition)
  24. Temp table caching with spark-sql
  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