Optimize the following code :"// Import the necessary Spark libraries
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

// Define the schema for the data feed
public class DataFeed {
    public String[] technicalFields;
    public String[] nonTechnicalFields;
}

// Define the schema for the record
public class Record {
    public String[] technicalFields;
    public String[] nonTechnicalFields;
}

public class HistorizationProcess {
    public static void main(String[] args) {
        // Create a Spark session
        SparkSession spark = SparkSession.builder()
            .appName("HistorizationProcess")
            .master("local[*]")
            .getOrCreate();

        // Create a Spark context
        JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());

        // Load the sourced data feed
        JavaRDD<DataFeed> sourcedDataFeed = sc.textFile("sourcedDataFeed.csv")
            .map(line -> {
                String[] parts = line.split(",");
                DataFeed dataFeed = new DataFeed();
                dataFeed.technicalFields = new String[]{ parts[0], parts[1], parts[2] };
                dataFeed.nonTechnicalFields = new String[]{ parts[3], parts[4], parts[5] };
                return dataFeed;
            });

        // Load the historical data feed
        JavaRDD<DataFeed> historicalDataFeed = sc.textFile("historicalDataFeed.csv")
            .map(line -> {
                String[] parts = line.split(",");
                DataFeed dataFeed = new DataFeed();
                dataFeed.technicalFields = new String[]{ parts[0], parts[1], parts[2] };
                dataFeed.nonTechnicalFields = new String[]{ parts[3], parts[4], parts[5] };
                return dataFeed;
            });

        // Convert the data feeds to key-value pairs
        JavaPairRDD<String, DataFeed> sourcedDataFeedPairs = sourcedDataFeed
            .mapToPair(dataFeed -> new Tuple2<>(String.join(",", dataFeed.technicalFields), dataFeed));
        JavaPairRDD<String, DataFeed> historicalDataFeedPairs = historicalDataFeed
            .mapToPair(dataFeed -> new Tuple2<>(String.join(",", dataFeed.technicalFields), dataFeed));

        // Compute the delta records
        JavaPairRDD<String, Record> deltaRecords = sourcedDataFeedPairs
            .leftOuterJoin(historicalDataFeedPairs)
            .filter(pair -> {
                DataFeed sourcedDataFeedRecord = pair._2()._1();
                DataFeed historicalDataFeedRecord = pair._2()._2().orElse(null);
                if (historicalDataFeedRecord == null) {
                    // The record is new
                    return true;
                } else if (!sourcedDataFeedRecord.nonTechnicalFields.equals(historicalDataFeedRecord.nonTechnicalFields)) {
                    // The record is updated
                    return true;
                } else {
                    // The record is constant
                    return false;
                }
            })
            .mapToPair(pair -> {
                DataFeed sourcedDataFeedRecord = pair._2()._1();
                DataFeed historicalDataFeedRecord = pair._2()._2().orElse(null);
                Record record = new Record();
                record.technicalFields = sourcedDataFeedRecord.technicalFields;
                record.nonTechnicalFields = sourcedDataFeedRecord.nonTechnicalFields;
                if (historicalDataFeedRecord != null) {
                    // The record is updated
                    record.nonTechnicalFields = historicalDataFeedRecord.nonTechnicalFields;
                }
                return new Tuple2<>(String.join(",", record.technicalFields), record);
            });

        // Union the delta records with the historical data feed
        JavaPairRDD<String, DataFeed> union = deltaRecords
            .mapToPair(record -> new Tuple2<>(record._1(), new DataFeed(record._2().technicalFields, record._2().nonTechnicalFields)))
            .union(historicalDataFeedPairs)
            .reduceByKey((record1, record2) -> record2);

        // Save the union to a file
        union.values().map(dataFeed -> String.join(",", dataFeed.technicalFields) + "," + String.join(",", dataFeed.nonTechnicalFields))
            .saveAsTextFile("historicalDataFeed_updated.csv");

        // Stop the Spark context
        sc.stop();
    }
}"