Delta Lake Compacting Несколько файлов в один файл

Question

В настоящее время я изучаю озеро дельты, которое открыто источниками данных. Я читаю данные Кафки и пишу как поток, используя формат дельты озера. Дельта-Лейк создает много файлов во время потоковой записи с kafka, которые, я чувствую, червы файловой системы hdfs.

Я попытался выполнить следующие действия, чтобы сжать несколько файлов в один файл.

val spark =  SparkSession.builder
    .master("local")
    .appName("spark session example")
    .getOrCreate()

  val df = spark.read.parquet("deltalakefile/data/")

  df.repartition(1).write.format("delta").mode("overwrite").save("deltalakefile/data/")
  df.show()

  spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled","false")

  DeltaTable.forPath("deltalakefile/data/").vacuum(1)

Но когда я проверил выводон создает новый файл, а не удаляет существующие.

Есть ли способ добиться этого. И какова здесь связь между сроком хранения? Как мы должны настроить его в HDFS при использовании? Какой должна быть моя конфигурация для хранения, когда я хочу создать необработанный / бронзовый слой в формате delta lake и хочу сохранить все свои данные в течение длительного периода (годы в помещениях / бесконечное время в облаке)?

Answer 1

По замыслу Delta не удаляет файлы сразу, чтобы предотвратить воздействие на активных потребителей. Он также обеспечивает управление версиями (так называемое путешествие во времени), так что вы можете просматривать историю при необходимости. Чтобы удалить предыдущие версии или незафиксированные файлы, вам нужно запустить vacu .

import io.delta.tables._

val deltaTable = DeltaTable.forPath(spark, pathToTable)

deltaTable.vacuum() // use default retention period

С точки зрения вашего вопроса о том, как управлять удержанием и уплотнением для модели бронза / серебро / золото, вам следуетРассматривайте свой посадочный стол (он же бронза) как бревно только для приложений. Это означает, что вам не нужно выполнять уплотнение или переписывать после факта. Бронзовая таблица должна представлять собой запись данных, которые вы приняли из вашего вышестоящего источника данных (например, Kafka) с минимальной обработкой.

Бронзовая таблица обычно используется в качестве дополнительного источника потока для заполнения нижестоящих наборов данных. Учитывая, что чтение из Delta выполняется из журнала транзакций, небольшие файлы не являются такой проблемой по сравнению со стандартными программами чтения файлов, которые выполняют медленные листинги.

Тем не менее, есть еще некоторые варианты для оптимизации файловкогда вы записываете их в бронзовую таблицу: 1) сжимайте ваши сообщения Kafka при записи в Delta, сначала перераспределяя их, чтобы уменьшить количество файлов, 2) увеличивайте интервал триггера, чтобы загрузка выполнялась реже и записывала больше сообщений в большуюфайлы.