Обработка дублирующихся записей при записи датафреймов Spark в базу данных Azure SQL Server с использованием Databricks

Условие задачи:

У нас есть задача в Databricks, состоящая из нескольких параллельно работающих задач. Каждая задача записывает Spark dataframe в таблицу Azure SQL Database. Каждая задача записывает в свою целевую таблицу. Некоторые записывают тысячи записей, но несколько из них — миллионы. Процесс является обрезкой-загрузкой из-за особенностей дизайна, также это усложняет возможность сделать его инкрементальным.

Этот процесс изначально работал нормально. Но после добавления некоторых загрузок больших таблиц мы время от времени сталкиваемся с дублирующимися строками в таблице SQL Server после записи. Мы проанализировали эти записи и поняли, что это связано с тем, что одна из партиций dataframe записывается дважды. Мы не знаем точную причину, но предполагаем, что это связано с проблемой связи между SQL Server и кластером Spark, когда SQL Server сообщает Spark о необходимости повторного выполнения задачи, тогда как записи уже были фактически записаны и зафиксированы. Потенциально это может быть связано с ограничениями со стороны Azure SQL Database, так как использование DTU достигает 100% во время загрузки.

Код для записи:

def write_to_sqldb_table(self, df: DataFrame, target_table_name: str, num_partitions: int = 1) -> None:

        df.write.format("SQLSERVER").option("host", self.sql_host_name).option("database", self.sql_db_name).option(
            "dbtable", target_table_name
        ).option("user", self.sql_user_id).option("password", self.sql_user_password).option("truncate", "true").option(
            "batchsize", self.batch_size
        ).option("numPartitions", num_partitions).mode("overwrite").save()

• batch_size установлен на 10.000, ранее мы проводили тестирование производительности, и это кажется оптимальным значением.
• num_partitions рассчитывается относительно количества записей в dataframe: total_records/batch_size, а затем округляется до ближайшего кратного 4, так как наш кластер имеет кратное 4 количество ядер.

Что мы пробовали до сих пор

Определение уникальных ограничений на SQL-таблице и обработка ошибок в PySpark

• Уникальные ограничения справились с поставленной задачей: заставили задачу Spark завершаться с ошибкой при вставке дублирующихся записей.
• Пробовали обработать ошибку в Spark, проверяя текст “Нарушение уникального ограничения” и затем возвращаясь из функции, но не удалось это реализовать. Задача Spark продолжала завершаться с ошибкой, вместо того чтобы поймать ошибку и продолжить с оставшимися вставками до успешного завершения задачи.

def write_to_sqldb_table_with_error_handling(self, df: DataFrame, target_table_name: str, num_partitions: int = 1) -> None:

        try:
            df.write.format("SQLSERVER").option("host", self.sql_host_name).option("database", self.sql_db_name).option(
                "dbtable", target_table_name
            ).option("user", self.sql_user_id).option("password", self.sql_user_password).option("truncate", "true").option(
                "batchsize", self.batch_size
            ).option("numPartitions", num_partitions).mode("overwrite").save()

        except Exception as e:
            error_message = str(e)

            if "Нарушение уникального ключевого ограничения" in error_message:
                return
            else:
                raise e

Определение уникальных ограничений на SQL-таблице и включение параметра IGNORE_DUP_KEY на уровне таблицы

• Операции вставки для крупных таблиц оказались заблокированными после внедрения IGNORE_DUP_KEY.
• Пробовали уменьшить batch_size, но это не помогло.

Запрос:

Мы ищем надежное решение для предотвращения дублирующих вставок при сохранении хорошей производительности. В данный момент я думаю о следующих двух решениях, которые стоит попробовать, но буду рад любым рекомендациям или опыту других, которые могут быть полезны!

• вариант 1: использование Apache Spark Connector для Azure SQL с reliabilityLevel = “NO_DUPLICATES”

  • этот коннектор часто обсуждается в контексте повышения производительности при записи Spark dataframe в базу данных Azure SQL
  • последний релиз поддерживает только Spark 3.4, в то время как мы работаем на Spark 3.5, но, думаю, стоит попробовать

• вариант 2: обработка дедупликации на стороне SQL Server, добавив столбец IDENTITY в качестве технического ключа в таблицы SQL Server и вызвав хранимую процедуру из моей задачи Spark для удаления дубликатов после операции вставки