Лучший способ обновить таблицу базы данных через задание pyspark

Question

У меня есть искровое задание, которое собирает данные из нескольких источников и объединяет их в одну таблицу. Задание должно обновлять таблицу только при наличии новых данных.

Один из подходов, который я мог придумать, - это извлечь данные из существующей таблицы и сравнить с новыми поступающими данными. Сравнение происходит в искровом слое.

Мне было интересно, есть ли лучший способ сравнения, который может улучшить производительность сравнения.

Пожалуйста, дайте мне знать, если у кого-то есть предложения по этому поводу.

Большое спасибо заранее.

Answer 1

Один из подходов, который я могу придумать, - это извлечь данные из существующей таблицы и сравнить с новыми данными, которые поступают в

ИМХО, сравнивать все данные для загрузки новых данных не Performant.

Вариант 1:

Вместо этого вы можете создать google-bigquery таблицу разделов и создать столбец раздела для загрузить данные, а также при загрузке новых данных вы можете проверить, есть ли у новых данных такой же столбец раздела. Удар по данным на уровне раздела в кусте или большом запросе более полезен / эффективен, чем выбор целых данных и сравнение в искре.

То же самое применимо и для улья.

см. Это Создание секционированных таблиц или Создание и использование секционированных таблиц с целочисленным диапазоном

---

Опция 2:

Другой альтернативой является с GOOGLE bigquery у нас есть оператор слияния, если вам нужно объединить данные без сравнения, тогда вы можете go вперед с оператором MERGE .. см. ссылку do c ниже Оператор MERGE - это оператор DML, который может объединять операции INSERT, UPDATE и DELETE в один оператор и выполнять операции атомарно.

Используя это, мы можем улучшить производительность, поскольку все три операции (INSERT, UPDATE и DELETE) выполняются за один проход. Нам не нужно писать отдельный оператор для обновления изменений в целевой таблице.

Answer 2

Существует много способов решения этой проблемы. Один из менее дорогих, производительных и масштабируемых способов - использовать хранилище данных в файловой системе для определения истинно новых данных.

1. По мере поступления данных в первый раз запишите его в 2 места - базу данных и в файл (скажем, в s3). Если данные уже есть в базе данных, вам нужно инициализировать локальный файл / s3 с табличными данными.
2. Поскольку данные поступают во второй раз и далее, проверьте, является ли оно новым, основываясь на его присутствии на локальном файле / s3.
3. Пометить дельта-данные как новые или обновленные. Экспортируйте это в базу данных как вставку или обновление.

Со временем этот файл будет становиться все больше и больше. Определите диапазон дат, после которого обновленные данные не будут поступать. Регулярно обрезайте этот файл, чтобы сохранить данные в этом диапазоне времени. Вы также можете разбить эти данные на сегменты. Вы можете использовать Deltalake, чтобы поддерживать его тоже.

Недостатком является то, что при каждом обновлении базы данных этот файл может нуждаться в обновлении на основании того, изменяются соответствующие данные или нет. Вы можете сохранить маркер в таблице базы данных для обозначения даты syn c. Индексируйте этот столбец тоже. Прочитайте измененные записи на основе этого столбца и обновите файл / deltalake.

Таким образом, ваше приложение sparl будет меньше зависеть от базы данных. Операции с базой данных не очень масштабируемы, поэтому лучше держать их подальше от критического пути

Answer 3

Разве вы не должны иметь время последнего обновления в вашей БД? Подход, который вы используете, не выглядит масштабируемым, поэтому, если у вас был способ установить время обновления для каждой строки в таблице, это решит проблему.