PySpark dataframe how to use flatmap

user_10033921 0
Viewed 345

I am writing a PySpark program that is comparing two tables, let's say Table1 and Table2 Both tables have identical structure, but may contain different data

Let's say, Table 1 has below cols

key1, key2, col1, col2, col3

The sample data in table 1 is as follows

"a", 1, "x1", "y1", "z1"
"a", 2, "x2", "y2", "z2"
"a", 3, "x3", "y3", "z3"

Similarly Table 2 has below cols

key1, key2, col1, col2, col3

The sample data in table 1 is as follows

"a", 1, "x1", "y1", "z1"
"a", 2, "x21", "y21", "z2"
"a", 3, "x3", "y3", "z31"

The program creates a data frame (let's say df1) that contains below columns

Key1, Key2, a.Col1, a.Col2, a.Col3, b.Col1, b.Col2, b.Col3, column_names

example data:

"a", 2, "x2", "y2", "z2", "x21", "y21", "z2", "col1,col2"
"a", 3, "x3", "y3", "z3", "x3", "y3", "z31", "col3"

The column "column_names" contains columns that have different values between table1 and table2

Using this data frame, I need to create another data frame that contains below structure

key1, key2, field_in_difference, src_value, tgt_value

"a", 2, "col1", "x2", "x21"
"a", 2, "col2", "y2", "y21"
"a", 3, "col3", "z3", "z31"

I am thinking that I need use flatMap in PySpark Can I use flatmap for one of the column in the dataframe, so that multiple rows are created in the resulting dataframe ? but remaining columns get copied in the new row ?

I tried to use following, but does not seem to be correct syntax

df2 = df1.withColumn("newcolumn", func.concat_ws(",", flatMap(lambda x: x.split(','))))

But I get an error NameErrorL name flatMap is not defined not sure how do I specify that the flatmap needs to be done on the column "column_names" , while keeping the remaining cols as they are..

I think the approach is to create one row per the column in difference as a first step Then in the second step, create another df that will transform as expected output

I really appreciate the help

pyspark
1 Answers

flatMap works on RDD, not DataFrame.

I don't quite understand how you want to use flatMap on df1, but I think working directly from Table 1 and Table 2 might be easier. Let's say Table 1 is df_src and Table 2 is df_tgt.

df_src.show()

+----+----+----+----+----+
|key1|key2|col1|col2|col3|
+----+----+----+----+----+
|   a|   1|  x1|  y1|  z1|
|   a|   2|  x2|  y2|  z2|
|   a|   3|  x3|  y3|  z3|
+----+----+----+----+----+

df_tgt.show()

+----+----+----+----+----+
|key1|key2|col1|col2|col3|
+----+----+----+----+----+
|   a|   1|  x1|  y1|  z1|
|   a|   2| x21| y21|  z2|
|   a|   3|  x3|  y3| z31|
+----+----+----+----+----+

You can un-pivot both dataframes using stack function, join them, and filter it.

from pyspark.sql.functions import col

# unpivot col1, col2 and col3 of both dataframes. rename key columns as well
df_src = df_src.selectExpr("key1 key1_s", "key2 key2_s", "stack(3, 'col1', col1, 'col2', col2, 'col3', col3) (field_s, src_value)")
df_tgt = df_tgt.selectExpr("key1 key1_t", "key2 key2_t", "stack(3, 'col1', col1, 'col2', col2, 'col3', col3) (field_t, tgt_value)")

# join the dataframes on keys and field, then filter where field values are different
df_res = (df_src
          .join(df_tgt, 
                [col('key1_s') == col('key1_t'), col('key2_s') == col('key2_t'), col('field_s') == col('field_t')], 
                'inner')
          .filter(col('src_value') != col('tgt_value'))
          .selectExpr('key1_s key1', 'key2_s key2', 'field_s field_in_difference', 'src_value', 'tgt_value')
          )
df_res.show()

+----+----+-------------------+---------+---------+
|key1|key2|field_in_difference|src_value|tgt_value|
+----+----+-------------------+---------+---------+
|   a|   2|               col1|       x2|      x21|
|   a|   2|               col2|       y2|      y21|
|   a|   3|               col3|       z3|      z31|
+----+----+-------------------+---------+---------+
user_94778430
Related