26
$\begingroup$

I have 10 data frames pyspark.sql.dataframe.DataFrame, obtained from randomSplit as (td1, td2, td3, td4, td5, td6, td7, td8, td9, td10) = td.randomSplit([.1, .1, .1, .1, .1, .1, .1, .1, .1, .1], seed = 100) Now I want to join 9 td's into a single data frame, how should I do that?

I have already tried with unionAll, but this function accepts only two arguments.

td1_2 = td1.unionAll(td2) 
# this is working fine

td1_2_3 = td1.unionAll(td2, td3) 
# error TypeError: unionAll() takes exactly 2 arguments (3 given)

Is there any way to combine more than two data frames row-wise?

The purpose of doing this is that I am doing 10-fold Cross Validation manually without using PySpark CrossValidator method, So taking 9 into training and 1 into test data and then I will repeat it for other combinations.

$\endgroup$
  • 1
    $\begingroup$ This does not directly answer the question, but here I give a suggestion to improve the naming method so that in the end, we don't have to type, for example: [td1, td2, td3, td4, td5, td6, td7, td8, td9, td10]. Imagine doing this for a 100-fold CV. Here's what I'll do: portions = [0.1]*10 cv = df7.randomSplit(portions) folds = list(range(10)) for i in range(10): test_data = cv[i] fold_no_i = folds[:i] + folds[i+1:] train_data = cv[fold_no_i[0]] for j in fold_no_i[1:]: train_data = train_data.union(cv[j]) $\endgroup$ – ngoc thoag Jun 26 '19 at 20:03
46
$\begingroup$

Stolen from: https://stackoverflow.com/questions/33743978/spark-union-of-multiple-rdds

Outside of chaining unions this is the only way to do it for DataFrames.

from functools import reduce  # For Python 3.x
from pyspark.sql import DataFrame

def unionAll(*dfs):
    return reduce(DataFrame.unionAll, dfs)

unionAll(td2, td3, td4, td5, td6, td7, td8, td9, td10)

What happens is that it takes all the objects that you passed as parameters and reduces them using unionAll (this reduce is from Python, not the Spark reduce although they work similarly) which eventually reduces it to one DataFrame.

If instead of DataFrames they are normal RDDs you can pass a list of them to the union function of your SparkContext

EDIT: For your purpose I propose a different method, since you would have to repeat this whole union 10 times for your different folds for crossvalidation, I would add labels for which fold a row belongs to and just filter your DataFrame for every fold based on the label

| improve this answer | |
$\endgroup$
  • $\begingroup$ (+1) A nice work-around. However, there needs to be a function which allows concatenation of multiple dataframes. Would be quite handy! $\endgroup$ – Dawny33 Apr 22 '16 at 8:39
  • $\begingroup$ I don't disagree with that $\endgroup$ – Jan van der Vegt Apr 22 '16 at 8:40
  • $\begingroup$ @JanvanderVegt Thanks, it works and the idea of adding labels to filter out the training and testing dataset, I did it already. Thank you very much for your help. $\endgroup$ – krishna Prasad Apr 23 '16 at 3:27
  • $\begingroup$ @Jan van der Vegt Can you please apply the same logic for Join and answer this question $\endgroup$ – GeorgeOfTheRF Jun 13 '17 at 10:50
  • $\begingroup$ stackoverflow.com/questions/44516409/… $\endgroup$ – GeorgeOfTheRF Jun 13 '17 at 10:50
10
$\begingroup$

Sometime, when the dataframes to combine do not have the same order of columns, it is better to df2.select(df1.columns) in order to ensure both df have the same column order before the union.

import functools 

def unionAll(dfs):
    return functools.reduce(lambda df1,df2: df1.union(df2.select(df1.columns)), dfs) 

Example:

df1 = spark.createDataFrame([[1,1],[2,2]],['a','b'])
# different column order. 
df2 = spark.createDataFrame([[3,333],[4,444]],['b','a']) 
df3 = spark.createDataFrame([555,5],[666,6]],['b','a']) 

unioned_df = unionAll([df1, df2, df3])
unioned_df.show() 

enter image description here

else it would generate the below result instead.

from functools import reduce  # For Python 3.x
from pyspark.sql import DataFrame

def unionAll(*dfs):
    return reduce(DataFrame.unionAll, dfs) 

unionAll(*[df1, df2, df3]).show()

enter image description here

| improve this answer | |
$\endgroup$
4
$\begingroup$

How about using recursion?

def union_all(dfs):
    if len(dfs) > 1:
        return dfs[0].unionAll(union_all(dfs[1:]))
    else:
        return dfs[0]

td = union_all([td1, td2, td3, td4, td5, td6, td7, td8, td9, td10])
| improve this answer | |
$\endgroup$
0
$\begingroup$

You could do it like this.

val myDFCsv = spark.read.format("csv")
   .option("sep","|")
   .option("inferSchema","true")
   .option("header","false")
   .load("mnt/rawdata/2019/01/01/client/ABC*.gz")

myDFCsv.show()
myDFCsv.head()
myDFCsv.count()


//////////////////////////////////////////
// If you also need to load the filename
val myDFCsv = spark.read.format("csv")
   .option("sep","|")
   .option("inferSchema","true")
   .option("header","false")
   .load("mnt/rawdata/2019/01/01/client/ABC*.gz")
   .withColumn("file_name",input_file_name())


myDFCsv.show(false)
myDFCsv.head()
myDFCsv.count()
| improve this answer | |
$\endgroup$
  • $\begingroup$ I have to construct mutiple queries , get each query dataset and union all of them how to it ? $\endgroup$ – BdEngineer Apr 26 at 15:26
  • $\begingroup$ You said you want to union files with the same schemas, right. The code that I showed you does exactly that. If it's not doing what you want, can you change your original post to show 'before' and 'after' examples of the data you are working with? $\endgroup$ – ASH Apr 26 at 15:43
  • $\begingroup$ any thought how to handle my case without for-loops .. at least i need to run it in parallel some how ... please advice... stackoverflow.com/questions/61391531/… $\endgroup$ – BdEngineer Apr 26 at 15:44
  • 1
    $\begingroup$ Well, Spark should be running in parallel, by default. I don't think you tried the code. Try it and you will see how it works. $\endgroup$ – ASH Apr 26 at 15:52
0
$\begingroup$
def unionAll(a,b):
    return a.unionByName(b)

sdf1_sdf2 = reduce(unionAll,[sdf1,sdf2])
| improve this answer | |
$\endgroup$
-1
$\begingroup$

I just did this:

union_data = td1.unionAll(td2).unionAll(td3).unionAll(td4)

worked like a charm

| improve this answer | |
$\endgroup$

Not the answer you're looking for? Browse other questions tagged or ask your own question.