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 oldData = spark.read().parquet(“path/to/oldData”);
        Dataset 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 joinedData = oldData.join(
                broadcast(newData),
                oldData.col(primaryKey).equalTo(newData.col(primaryKey)),
                “fullouter”
        ).withColumn(“equalHash”, oldData.col(hashColumn).equalTo(newData.col(hashColumn)));

        Dataset unchangedData = joinedData
                .filter(joinedData.col(“equalHash”))
                .select(oldData.columns());

        Dataset changedOldData = joinedData
                .filter(joinedData.col(“equalHash”).notEqual(true))
                .select(oldData.columns())
                .withColumn(“end_date”, current_date())
                .withColumn(“current_flag”, lit(false));

        Dataset changedNewData = joinedData
                .filter(joinedData.col(“equalHash”).notEqual(true))
                .select(newData.columns())
                .withColumn(“start_date”, current_date())
                .withColumn(“end_date”, lit(null).cast(“date”))
                .withColumn(“current_flag”, lit(true));

        Dataset newRows = newData.join(
                broadcast(oldData),
                newData.col(primaryKey).equalTo(oldData.col(primaryKey)),
                “leftanti”
        )
                .withColumn(“start_date”, current_date())
                .withColumn(“end_date”, lit(null).cast(“date”))
                .withColumn(“current_flag”, lit(true));

        Dataset result = unchangedData
                .union(changedOldData)
                .union(changedNewData)
                .union(newRows);

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

    private static Dataset createHash(Dataset data, String[] nonKeyColumns, String hashColumn) {
        Dataset 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));
    }
}"