Can I get metadata of files reading by Spark

Question

Let's suppose we have 2 files, file#1 created at 12:55 and file#2 created at 12:58. While reading these two files I want to add a new column "creation_time". Rows belong to file#1 have 12:55 in "creation_time" column and Rows belong to file#2 have 12:58 in "creation_time".

new_data = spark.read.option("header", "true").csv("s3://bucket7838-1/input")

I'm using above code snippet to read the files in "input" directory.

Answer 1

Use input_file_name() function to get the filename and then use hdfs file api to get the file timestamp finally join both dataframes on filename.

Example:

from pyspark.sql.types import *
from pyspark.sql.functions import *
URI           = sc._gateway.jvm.java.net.URI
Path          = sc._gateway.jvm.org.apache.hadoop.fs.Path
FileSystem    = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem
Configuration = sc._gateway.jvm.org.apache.hadoop.conf.Configuration

fs = FileSystem.get(URI("hdfs://<namenode_address>:8020"), Configuration())

status = fs.listStatus(Path('<hdfs_directory>'))

filestatus_df=spark.createDataFrame([[str(i.getPath()),i.getModificationTime()/1000] for i in status],["filename","modified_time"]).\
withColumn("modified_time",to_timestamp(col("modified_time")))

input_df=spark.read.csv("<hdfs_directory>").\
withColumn("filename",input_file_name())

#join both dataframes on filename to get filetimestamp
df=input_df.join(filestatus_df,['filename'],"left")

Answer 2

Here are the steps

1. Use sparkcontext.wholeTextFiles("/path/to/folder/containing/all/files")
2. The above returns an RDD where key is the path of the file, and value is the content of the file
3. rdd.map(lambda x:x[1]) - this give you an rdd with only file contents
4. rdd.map(lambda x: customeFunctionToProcessFileContent(x))
5. since map function works in parallel, any operations you do, would be faster and not sequential - as long as your tasks don't depend on each other, which is the main criteria for parallelism

import os
import time

import pyspark
from pyspark.sql.functions import udf
from pyspark.sql.types import *

# reading all the files to create PairRDD 
input_rdd = sc.wholeTextFiles("file:///home/user/datatest/*",2)

#convert RDD to DF

input_df=spark.createDataFrame(input_rdd)

input_df.show(truncate=False)
'''
+---------------------------------------+------------+
|_1                                     |_2          |
+---------------------------------------+------------+
|file:/home/user/datatest/test.txt      |1,2,3  1,2,3|
|file:/home/user/datatest/test.txt1     |4,5,6  6,7,6|
+---------------------------------------+------------+
'''
input_df.select("_2").take(2)
#[Row(_2=u'1,2,3\n1,2,3\n'), Row(_2=u'4,5,6\n6,7,6\n')]


# function to get a creation time of a file
def time_convesion(filename):
    return time.ctime(os.path.getmtime(filename.split(":")[1]))

#udf registration
time_convesion_udf = udf(time_convesion, StringType())

#udf apply over the DF
final_df = input_df.withColumn("created_time", time_convesion_udf(input_df['_1']))

final_df.show(2,truncate=False)
'''
+---------------------------------------+------------+------------------------+
|_1                                     |_2          |created_time            |
+---------------------------------------+------------+------------------------+
|file:/home/user/datatest/test.txt      |1,2,3  1,2,3|Sat Jul 11 18:31:03 2020|
|file:/home/user/datatest/test.txt1     |4,5,6  6,7,6|Sat Jul 11 18:32:43 2020|
+---------------------------------------+------------+------------------------+
'''
# proceed with the next steps for the implementation

The above works with default partition though. So you might not get input files count equal to output file count(as output is number of partitions).

You can re-partition the RDD based on count or any other unique value based on your data, so you end up with output files count equal to input count. This approach will have only parallelism but will not have the performance achieved with optimal number of partitions