Optimize the following code by reducing the number of joins : "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 unchangedData = oldData.join(
broadcast(newData),
oldData.col(primaryKey).equalTo(newData.col(primaryKey))
)
.where(oldData.col(hashColumn).equalTo(newData.col(hashColumn)))
.select(oldData.columns());

Dataset changedOldData = oldData.join(
broadcast(newData),
oldData.col(primaryKey).equalTo(newData.col(primaryKey))
)
.where(oldData.col(hashColumn).notEqual(newData.col(hashColumn)))
.select(oldData.columns())
.withColumn(“end_date”, current_date())
.withColumn(“current_flag”, lit(false));

Dataset changedNewData = newData.join(
broadcast(oldData),
newData.col(primaryKey).equalTo(oldData.col(primaryKey))
)
.where(newData.col(hashColumn).notEqual(oldData.col(hashColumn)))
.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));
}
}"