Its really good how you explained the problem. I ran into similar issue with too many parquet files & too much time to write or stages hanging in the middle when i have to create dynamic columns (more than 1000) and write atleast 10M rows to S3.

One mistake i was making was i was doing all the operations in RDD instead of dataframe something like sqlContext().toJavaRDD or .rdd(). You lose lot of the good things once you move to rdd from dataframe. The below approach kind of worked for me, try it out and let me know

You can split your spark job into two parts:

1. Run a reduce() job to determine the dynamic column names that is required for storing the data

2. Run a flatMap() to transform the data that you received and convert them in to the schema that was created using the dynamic columns in step 1. Once that is done you can directly write them out as delta. Since delta is transactional based and does lot of book keeping you will not end up with too many parquet files.

Here is a sample of the step that i explained above:

1. // Part 1 --> Determine the dynamic column names
2. Row schemaOutput = testDataSet.reduce((ReduceFunction<Row>)(v1, v2)->{
3. List<String> set1, set2;
4. String data = v1.size()> OUTPUT_COLUMN ? v1.getString(OUTPUT_COLUMN):"";
5. if(data.startsWith("{"))
6. set1 =ScoringTransformer.getColumns(data);
7. elseset1 =newLinkedList<>((Collection<String>)(Collection<?>)JavaConverters.asJavaCollectionConverter(v1.toSeq()).asJavaCollection());
8. data = v2.size()> OUTPUT_COLUMN ? v2.getString(OUTPUT_COLUMN):"";
9. if(data.startsWith("{"))
10. set2 =ScoringTransformer.getColumns(data);
11. elseset2 =newLinkedList<>((Collection<String>)(Collection<?>)JavaConverters.asJavaCollectionConverter(v2.toSeq()).asJavaCollection());
12. Set<String> columns =newLinkedHashSet<>(set1);
13. columns.addAll(set2);
14. returnRowFactory.create(columns.toArray());
15. });

1. List<String> columns =newLinkedList<>((Collection<String>)(Collection<?>)
2. JavaConverters.asJavaCollectionConverter(schemaOutput.toSeq()).asJavaCollection());
3. // You get a schema out for all the dynamic column names that is required for the next step
4. ExpressionEncoder<Row> encoder =RowEncoder.apply(ReportUtils.getSchemaBasedOnString(columns));

1. // Part 2 --> Run Flatmap and encode it with the schema that you created out of dynamic columns, then repartition in whatever way you want and write it out to s3
2. testDataSet.flatMap((FlatMapFunction<Row,Row>) row ->{
3. List<Map<String,String>> data =ScoringTransformer.getTransformedData(row.getString(OUTPUT_COLUMN));
4. if(data !=null&&!data.isEmpty())
5. return data.stream()
6. .map(x ->RowFactory.create(columns.parallelStream()
7. .map(y -> x.getOrDefault(y,""))
8. .collect(Collectors.toList()).toArray()))
9. .collect(Collectors.toList())
10. .iterator();
11. returnnewLinkedList<Row>().iterator();
12. }, encoder)
13. .repartition(col("date_group"))
14. .write()
15. .option("mergeSchema","true")
16. .format(FORMAT)
17. .mode("append")
18. .save(s3Output +"/xyz");