Конкатенация улья, аналог в искре

Question

У меня есть искровое задание, которое создает одну таблицу результатов в Hive и загружает данные из других таблиц. Эта работа выполняется каждый день и генерирует много файлов в HDFS. Когда я использовал Hive, я создал скрипт: alter table quality_of_service_1 concatenate;

implicit val spark: SparkSession = SparkSession
    .builder()
    .enableHiveSupport()
    .appName("Test")
    .getOrCreate()


  def build(implicit session: SparkSession, config: Config): Unit = {
    import session.implicits._
    loadData
      .flatMap(Item.buildInternal(_, config))
      .write
      .mode(SaveMode.Overwrite)
      .format("orc")
      .saveAsTable(s"${config.schema}.result_table")
  }

Не могли бы вы написать, как объединить эти файлы и где они настраиваются? Скрипт оболочки с HDFS, спаржи или спаржи сессионных конфигов.

Конфиг:

--deploy-mode cluster \
--conf spark.rpc.message.maxSize=300 \
--conf spark.rdd.compress=true \
--conf spark.default.parallelism=1009 \
--conf spark.sql.shuffle.partitions=1009 \
--conf spark.sql.autoBroadcastJoinThreshold=31457280 \
--conf spark.dynamicAllocation.enabled=true \
--conf spark.shuffle.service.enabled=true \
--conf spark.dynamicAllocation.initialExecutors=1200 \
--conf spark.dynamicAllocation.minExecutors=400 \
--conf spark.dynamicAllocation.maxExecutors=1200 \
--conf spark.sql.files.maxPartitionBytes=1073741824 \
--executor-cores 3 \
--executor-memory 7g \
--driver-memory 4g \

Answer 1

В spark2 параметр контролирует размер одного раздела ... и, следовательно, количество файлов, когда вы «сохраняете». Поэтому увеличьте параметр, чтобы иметь меньше разделов и меньше файлов. Например, один гигабайт на раздел val maxSplit=1024*1024*1024 spark.conf.set("spark.sql.files.maxPartitionBytes", maxSplit)