@Service
public class EtlService {
    private static final Logger logger = LoggerFactory.getLogger(EtlService.class);

    @Autowired
    private NamedParameterJdbcTemplate jdbcTemplate;

    @Autowired
    private Environment env;

    @Autowired
    private EntityManager em;

    private HashMap<String, Long> watermarks;

    public void copyMysqlTable(SparkSession spark, String tableName, String idColumn) {
        logger.info("copyMysqlTable: copying table {}",tableName);
        long tableWatermark = loadWatermark(tableName) ;

        Dataset<Row> mysqlDf = spark.read()
                .format("jdbc")
                .option("url", env.getProperty("mysql.datasource.url"))
                .option("dbtable", String.join(".","mydwhdwh",tableName))
                .option("user", env.getProperty("mysql.datasource.username"))
                .option("password", env.getProperty("mysql.datasource.password"))
                .load();

        mysqlDf = mysqlDf.filter(mysqlDf.col("start_trxid").gt(tableWatermark));

        mysqlDf.write()
                .format("jdbc")
                .mode("overwrite")
//                .option("truncate", true)
                .option("url", env.getProperty("spring.datasource.url"))
                .option("dbtable", tableName+"_staging")
                .option("user", env.getProperty("spring.datasource.username"))
                .option("password", env.getProperty("spring.datasource.password"))
                .save();

        List<String> columnList = new ArrayList();
        Arrays.stream(mysqlDf.schema().fields()).forEach(x -> columnList.add(x.name()));

        executeMergeStatement(columnList, tableName, idColumn);

    }

    private void executeMergeStatement(List<String> columnList, String tableName, String idColumn) {
        var setList = columnList.stream().map(s-> "t."+s+" = s."+s).collect(Collectors.joining(", "));
        var valList = columnList.stream().map(s-> "s."+s).collect(Collectors.joining(", "));

        var query = """
		MERGE $target AS t
		USING $source	AS s
		ON (t.$idcolumn = s.$idcolumn)
		WHEN MATCHED
			 THEN UPDATE
			 SET $setlist         
		WHEN NOT MATCHED BY Target THEN
		INSERT ($columnlist)
		VALUES ($vallist);
		""".replace("$target", tableName)
            .replace("$source", tableName+"_staging")
            .replace("$columnlist", String.join(",", columnList))
            .replace("$idcolumn", idColumn)
            .replace("$setlist", setList)
            .replace("$vallist", valList);


        var namedParameters = new MapSqlParameterSource();
        jdbcTemplate.update(query,namedParameters);
    }
...