Optimize the following code : "irst, we can merge the three join operations that use the same join condition, making use of the when() clause to create or update columns accordingly. This allows us to make just one join operation instead of three separate ones.

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;

import static org.apache.spark.sql.functions.*;

public class SCDType2Example {
    public static void main(String[] args) {
        SparkSession spark = SparkSession
                .builder()
                .appName(“SCDDevice2Example”)
                .master(“local”)
                .getOrCreate();

        Dataset<Row> oldData = spark.read().parquet(“path/to/oldData”);
        Dataset<Row> newData = spark.read().parquet(“path/to/newData”);

        // Repartition the data
        oldData = oldData.repartition(“id”);
        newData = newData.repartition(“id”);

        String primaryKey = “id”;
        String[] nonKeyColumns = {“name”, “age”, “city”}; // Parametrizable columns
        String hashColumn = “hash”;

        oldData = createHash(oldData, nonKeyColumns, hashColumn);
        newData = createHash(newData, nonKeyColumns, hashColumn);

        Dataset<Row> joinedData = oldData.join(
                broadcast(newData),
                oldData.col(primaryKey).equalTo(newData.col(primaryKey)),
                “fullouter” // Changed
        );

        Dataset<Row> result = joinedData
                .withColumn(“hash_1”, coalesce(oldData.col(“hash”), lit(“”)))
                .withColumn(“hash_2”, coalesce(newData.col(“hash”), lit(“”)))
                .withColumn(“date”, current_date())
                .select(
                        when(oldData.col(primaryKey).isNull(), newData.col(primaryKey)).otherwise(oldData.col(primaryKey)).alias(primaryKey),
                        when(oldData.col(“hash”).equalTo(newData.col(“hash”)), oldData.col(“name”)).otherwise(newData.col(“name”)).alias(“name”),
                        when(oldData.col(“hash”).equalTo(newData.col(“hash”)), oldData.col(“age”)).otherwise(newData.col(“age”)).alias(“age”),
                        when(oldData.col(“hash”).equalTo(newData.col(“hash”)), oldData.col(“city”)).otherwise(newData.col(“city”)).alias(“city”),
                        when(oldData.col(“hash”).equalTo(newData.col(“hash”)), oldData.col(“start_date”)).otherwise(oldData.col(“start_date”)).alias(“start_date”),
                        when(oldData.col(“hash”).notEqual(newData.col(“hash”)), current_date()).otherwise(oldData.col(“end_date”)).alias(“end_date”),
                        when(oldData.col(“hash”).isNull(), lit(true)).otherwise(oldData.col(“hash”).equalTo(newData.col(“hash”))).alias(“current_flag”),
                        when(newData.col(“hash”).isNull(), lit(false)).otherwise(oldData.col(“hash”).notEqual(newData.col(“hash”))).alias(“update_flag”)
                );

        result.write().parquet(“path/to/output”);
    }

    private static Dataset<Row> createHash(Dataset<Row> data, String[] nonKeyColumns, String hashColumn) {
        Dataset<Row> result = data;
        StringBuilder concatString = new StringBuilder();

        for (String column : nonKeyColumns) {
            concatString.append(“cast(”).append(column).append(" as string),“);
        }

        String concatColumns = concatString.toString();

        if (concatColumns.endsWith(”,“)) {
            concatColumns = concatColumns.substring(0, concatColumns.length() - 1);
        }

        return result.withColumn(hashColumn, sha2(functions.concat_ws(”", functions.expr(concatColumns)), 256));
    }
}"