Обработка специальных символов и escape-последовательностей в CSV с помощью команды Spark и PostgreSQL COPY ⇐ Python

[Anonymous]

Я использую PySpark для перераспределения DataFrame и записи его в файл CSV, который позже загружаю в PostgreSQL с помощью команды COPY. Однако при выполнении команды COPY я столкнулся со следующей ошибкой:
psycopg2.errors.BadCopyFileFormat: extra data after last expected column

Я подозреваю, что эта ошибка вызвана специальными символами в моих данных, например двойными кавычками ("), запятые (,), косая черта (/) и escape-символы. Вот упрощенная версия моего кода:
df_ops.df.repartition(num_partitions).write.mode("overwrite").format("csv").option("header", "false").save(file_path)

with connection.cursor() as cursor:
with open(file_path, "r") as f:
query = f"COPY {table_name} ({column_string}) FROM STDIN WITH (FORMAT CSV, HEADER FALSE, DELIMITER ',', NULL '')"
cursor.copy_expert(query, f)
connection.commit()

Пример данных:
Исходная строка:
2024-07-18 12:09:28,2012420012080452979,60,V2,"Suspected fraud, decline / Transactions declined based on Risk Score",1

После преобразования (CSV):
2024-07-18 12:09:28,2012420012080452979,60,V2,"\"Suspected fraud, decline / Transactions declined based on Risk Score\"",1

Проблема:

• Мои данные содержат смесь специальных символов, запятых и двойных кавычек.
• Команда COPY в PostgreSQL выдает ошибку при обработке CSV из-за этих специальных символов.

Вопрос:
Как я могу гарантировать, что вывод CSV из Spark правильно отформатирован для обработки всех специальных символов (двойных кавычек, запятых, косой черты и т. д.) при использовании команды PostgreSQL COPY ? Должен ли я настроить параметры записи Spark или оператор PostgreSQL COPY, чтобы справиться с этим?
Любые рекомендации о том, как обрабатывать эти escape-последовательности и обеспечивать обработку команд COPY были бы очень признательны за правильность данных.
Обновление:
Я добавил опцию цитирования, которая решила проблему; однако меня беспокоят возможные побочные эффекты.
df_ops.df.repartition(num_partitions).write.mode("overwrite").format("csv").option("header", "false").option("quote", "\u001C").save(file_path)
query = f"COPY {table_name} ({column_string}) FROM STDIN WITH (FORMAT CSV, HEADER FALSE, DELIMITER ',', NULL '', QUOTE '\u001C')"


Подробнее здесь: https://stackoverflow.com/questions/790 ... -postgresq