Optimize the following code : "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));
    }
}"