Efficient way of appending Spark DataFrames in a loop using pyspark

user_13107596 0
Viewed 4240

I have '|' delimited huge text files, I want to merge all the text files and create one huge spark dataframe, it will be later used for ETL process, using pyspark.

Inefficient way

1) Create an empty spark dataframe, df

2) In a loop,read the text file as to spark dataframe df1 and appending it to empty spark dataframe df

df = spark.createDataFrame([],schema)
for x in os.listdir(textfiles_dir):
    filepath = '{}/{}'.format(textfiles_dir,x)
    df1 = spark.read.format("csv") \
                    .option("header", "true") \
                    .option("delimiter", "|") \
                    .option("inferSchema","true") \
                    .load(filepath)
    df = df.union(df1)

This is not an efficient spark way.

Can anyone suggest an efficient way of doing this? That would be great if explained with sample code.

Thanks :)

pyspark-sql pyspark-dataframes pyspark apache-spark-sql apache-spark
3 Answers

filepath = filepath of directory where multiple files exists

dataframe = spark.read.format("csv").option("header", "true").option("delimiter", "|").load(filepath )

user_131075960
  1. df1 = spark.read. ... .load("pathFolder/") - read all files with folder
  2. df1 save as table db or file
user_117496510

As others have pointed out, you will want to read the entire text files directory as a dataframe, rather than iteratively reading each individual directory:

df = spark.read.format("csv") \
                    .option("header", "true") \
                    .option("delimiter", "|") \
                    .option("inferSchema","true") \
                    .load(textfiles_dir)

If you really want to go the union route, I would recommend using the union function in the SparkContext (http://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=union#pyspark.SparkContext.union) instead of the union function in DataFrame:

dfs = []
for x in os.listdir(textfiles_dir):
   filepath = '{}/{}'.format(textfiles_dir,x)
   df1 = spark.read.format("csv") \
                .option("header", "true") \
                .option("delimiter", "|") \
                .option("inferSchema","true") \
                .load(filepath)
   dfs.append(df1)
df = spark.sparkContext.union(dfs)
user_130736770
Related