PySpark Reading Multiple Files in Parallel

I had a similar problem, and it seems that I found a way: 1. Get a list of files 2.parallelize this list (distribute among all nodes) 3.write a function that reads content of all files from the portion of the big list that was distributed to the node 4.run it with mapPartition, then collect the result as a list, each element is a collected content of each file. Fot files stored on AWS s3 and json files:

def read_files_from_list(file_list):
#reads files from  list
#returns content as list of strings, 1 json per string ['{}','{}',...]
   out=[]
   for x in file_list:
      content = sp.check_output([ 'aws', 's3', 'cp', x, '-']) # content of the file. x here is a full path: 's3://bucket/folder/1.json'
      out.append(content)   
   return out #content of all files from the file_list as list of strings, 1 json per string ['{}','{}',...]


file_list=['f1.json','f2.json',...]
    ps3="s3://bucket/folder/"
    full_path_chunk=[ps3 + f for f in file_list] #makes list  of strings, with full path for each file
    n_parts = 100
    rdd1 = sc.parallelize(full_path_chunk, n_parts ) #distribute files among nodes
    list_of_json_strings = rdd1.mapPartitions(read_files_from_list).collect()

Then, if necessary, you can create spark dataframe like this:

rdd2=sc.parallelize(list_of_json_strings) #this is a trick! via http://spark.apache.org/docs/latest/sql-programming-guide.html#json-datasets
df_spark=sqlContext.read.json(rdd2)

The function read_files_from_list is just an example, it should be changed to read files from hdfs using python tools. Hope this helps :)

You can put all input files in the same directory then you can pass path of directory to spark. You can also use globbing like /data_dir/*.csv.

I had encountered similar situation recently. You can pass a list of CSVs with their paths to spark read api like spark.read.json(input_file_paths) (source). This will load all the files in a single dataframe and all the transformations eventually performed will be done in parallel by multiple executors depending on your spark config.