apache spark using Pyspark ETL help

Basically I have an ETL with 2 updates and I want to write the same updates in Pyspark table_a: +---+-----------+-------+--------------+ |key|col_a | col_b | current_flag | +---+-----------+-------+--------------+ |001| Value1 | T123 | Y | |002| oth_val1 | T123 | N | |003| oth_val2 | T123 | N | |004| oth_val3 | T123 | N | |005| Value2 | T123 | Y | |006| oth_val4 | T789 | N | |007| Value2 | T789 | Y | |008| Value1 | T789 | N | +---+-----------+-------+--------------+ UPDATE table_abc SET col_a = 'Value1' WHERE col_b IN ( SELECT col_b FROM table_abc WHERE col_a = 'Value1' and current_flag = 'Y' ) AND current_flag = 'N' COMMIT; +---+-----------+-------+--------------+ |key|col_a | col_b | current_flag | +---+-----------+-------+--------------+ |001| Value1 | T123 | Y | |002| Value1 | T123 | N | -- updated |003| Value1 | T123 | N | -- updated |004| Value1 | T123 | N | -- updated |005| Value2 | T123 | Y | |006| oth_val4 | T789 | N | |007| Value2 | T789 | Y | |008| Value1 | T789 | N | +---+-----------+-------+--------------+ UPDATE table_abc SET col_a = 'Value2' WHERE col_b IN ( SELECT col_b FROM table_abc WHERE col_a = 'Value2' and current_flag = 'Y' ) AND current_flag = 'N' COMMIT +---+-----------+-------+--------------+ |key|col_a | col_b | current_flag | +---+-----------+-------+--------------+ |001| Value1 | T123 | Y | |002| Value1 | T123 | N | |003| Value1 | T123 | N | |004| Value1 | T123 | N | |005| Value2 | T123 | Y | |006| Value2 | T789 | N | -- updated |007| Value2 | T789 | Y | |008| Value2 | T789 | N | -- updated +---+-----------+-------+--------------+ --------------------------------------------------------- #pyspark code to reproduce the updates #initial dataframe is "table_a" tval1 = [login to view URL]( col("col_a") == lit("Value1") & col("current_flag") == lit("Y") ) t= [login to view URL]("t1").join( [login to view URL]("tval1"), col("t1.col_b") == col("tval1.col_b"), "left-outer" ).select( col("[login to view URL]"), when( col("tval1.col_b").isNotNull(), lit("Value1") ).otherwise(col("t1.col_a")).alias("col_a"), col("t1.col_b"), col("t1.current_flag") ) #use data frame t from above tval2 = [login to view URL]( col("col_a") == lit("Value2") & col("current_flag") == lit("Y") ) t_new = [login to view URL]("t1").join( [login to view URL]("tval2"), col("t1.col_b") == col("tval2.col_b"), "left-outer" ).select( col("[login to view URL]"), when( col("tval2.col_b").isNotNull(), lit("Value2") ).otherwise(col("t1.col_a")).alias("col_a"), col("t1.col_b"), col("t1.current_flag") ) but what really happens in Pyspark is this: t_new: +---+-----------+-------+--------------+ |key|col_a | col_b | current_flag | +---+-----------+-------+--------------+ |001| Value1 | T123 | Y | |002| Value2 | T123 | N | |003| Value2 | T123 | N | |004| Value2 | T123 | N | |005| Value2 | T123 | Y | |006| Value2 | T789 | N | |007| Value2 | T789 | Y | |008| Value2 | T789 | N | +---+-----------+-------+--------------+