How to merge two parquet files having different schema in spark (java)

You can use mergeSchema option along with adding all the paths of parquet files you want to merge in parquet method, as follow:

Dataset<Row> finalDataset = testSparkSession.read()
  .option("mergeSchema", true)
  .parquet("D:\\ABC\\abc.parquet", "D:\\EFG\\efg.parquet");

All columns present in first dataset but not in second dataset will be set with null value in the second dataset

To merge two rows that come from two different dataframes, you first join the two dataframes, then select the right columns according on how you want to merge.

So for your case, it means:

• Read separately the two dataframes from their parquet location
• Join the two dataframes on their epochTime column, using a full_outer join as you want to keep all rows present in one dataframe but not in the other
• From the new dataframe with all the columns of the two dataframes duplicated, select merged columns using a function columnMerges (implementation below)
• [Optional] Reorder final dataframe by epochTime

Translated into code:

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

Dataset<Row> dataframe1 = testSparkSession.read().parquet("D:\\ABC\\abc.parquet");
Dataset<Row> dataframe2 = testSparkSession.read().parquet("D:\\EFG\\efg.parquet");

dataframe1.join(dataframe2, dataframe1.col("epochTime").equalTo(dataframe2.col("epochTime")), "full_outer")
  .select(Selector.columnMerges(dataframe2, dataframe1))
  .orderBy("epochTime")

Note: when we read parquets no need for mergeSchema option as for each dataframe we read only one parquet file thus only one schema

For the merge function Selector.columnMerges, for each row, what we want to do is:

• if the column is present in both dataframe, take value in dataframe2 if not null, else take value in dataframe1
• if the column is only present in dataframe2, take value in dataframe2
• if the column is only present in dataframe1, take value in dataframe1

So we first build set of columns of dataframe1, set of columns of dataframe2, and the list of columns from both dataframes, deduplicated. Then we iterate over this list of columns, applying previous rules for each one:

import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

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

public class Selector {

  public static Column[] columnMerges(Dataset<Row> main, Dataset<Row> second) {
    List<Column> columns = new ArrayList<>();

    Set<String> columnsFromMain = new HashSet<>(Arrays.asList(main.columns()));
    Set<String> columnsFromSecond = new HashSet<>(Arrays.asList(second.columns()));

    List<String> columnNames = new ArrayList<>(Arrays.asList(main.columns()));
    for (String column: second.columns()) {
      if (!columnsFromMain.contains(column)) {
        columnNames.add(column);
      }
    }

    for (String column : columnNames) {
      if (columnsFromMain.contains(column) && columnsFromSecond.contains(column)) {
        columns.add(when(main.col(column).isNull(), second.col(column)).otherwise(main.col(column)).as(column));
      } else if (columnsFromMain.contains(column)) {
        columns.add(main.col(column).as(column));
      } else {
        columns.add(second.col(column).as(column));
      }
    }

    return columns.toArray(new Column[0]);
  }
}