Решение проблем с форматами хранения данных

Симптомы

• PyArrow отказывается конвертировать pandas DataFrame с datetime64[ns] в Parquet
• Ошибка при записи DataFrame с наносекундными timestamp в Parquet через PyArrow
• pandas to_parquet() падает с ошибкой casting timestamp

Причина

Parquet хранит timestamp с микросекундной точностью (по умолчанию), а pandas использует наносекунды (datetime64[ns]). PyArrow 13+ по умолчанию запрещает потерю точности при downcast.

Решение

1. Используйте coerce_timestamps='us' при записи: pq.write_table(table, 'file.parquet', coerce_timestamps='us')
2. Или конвертируйте до записи: df['ts'] = df['ts'].dt.floor('us')
3. Для сохранения ns: pq.write_table(table, 'file.parquet', coerce_timestamps='ns', allow_truncated_timestamps=False)

Связанные уроки:

Симптомы

• PyArrow/DuckDB не может открыть .parquet файл
• Файл создан Spark, но не читается локально
• Ошибка при чтении файла после незавершённой записи

Причина

Файл повреждён или не является Parquet. Parquet-файл должен начинаться и заканчиваться magic bytes PAR1 (4 байта). Частые причины: неполная запись (writer упал до flush footer), файл переименован из другого формата, или это _SUCCESS/_metadata файл.

Решение

1. Проверьте magic bytes: head -c 4 file.parquet | xxd — должно быть PAR1
2. Если Spark-выход — пропустите служебные файлы: pq.read_table('dir/', filters=...) вместо чтения конкретного файла
3. Для повреждённых файлов: используйте parquet-tools meta file.parquet для диагностики
4. Убедитесь, что writer закрыл файл: writer.close() или context manager with pq.ParquetWriter(...)

Связанные уроки:

Симптомы

• Запрос к Parquet-файлу не находит колонку, которая точно была записана
• DuckDB/Spark показывают другие имена колонок, чем ожидалось
• Schema merge в Spark возвращает неожиданные типы или имена

Причина

Имя колонки отличается из-за case sensitivity (Parquet хранит имена as-is), автоматического переименования или использования разных схем записи. В Spark schema merging может объединить файлы с разными схемами некорректно.

Решение

1. Проверьте реальную схему файла: pq.read_schema('file.parquet') или DESCRIBE SELECT * FROM 'file.parquet' в DuckDB
2. В Spark включите case-insensitive чтение: spark.conf.set('spark.sql.caseSensitive', 'false')
3. При schema merge укажите schema явно: .option('mergeSchema', 'true') или .schema(explicit_schema)
4. Используйте pq.read_table('file.parquet', columns=['col1','col2']) с точными именами

Связанные уроки:

Симптомы

• Spark task падает с OOM при записи Parquet с Dictionary Encoding
• Memory footprint растёт линейно с количеством уникальных значений в колонке
• Запись с сортировкой потребляет больше памяти, чем ожидалось

Причина

Dictionary Encoding накапливает все уникальные значения в памяти до записи Data Page. Если кардинальность колонки высокая (URL, UUID), словарь занимает гигабайты. Parquet переключается на PLAIN при переполнении страницы, но к этому моменту словарь уже в памяти.

Решение

1. Отключите dictionary для high-cardinality колонок: .option('parquet.enable.dictionary', 'false') или pq.write_table(table, 'out.parquet', use_dictionary=['low_card_col'])
2. Уменьшите row_group_size для снижения пика памяти
3. В Spark: spark.conf.set('spark.sql.parquet.columnarWriterBatchSize', '1024') для уменьшения пакетов записи
4. Рассмотрите DELTA_BYTE_ARRAY для строк с общими префиксами (URL, пути)

Связанные уроки:

Симптомы

• Десериализация Avro-сообщения из Kafka падает с AvroTypeException
• Consumer не может прочитать данные, записанные с новой версией схемы
• Schema resolution между writer и reader schema завершается ошибкой

Причина

Reader schema несовместима с writer schema. Типичные причины: поле добавлено без default-значения (нарушение backward compatibility), nullable поле читается как required, или тип данных изменён несовместимо (string → int).

Решение

1. Проверьте совместимость схем: curl -X POST -d '{"schema": ...}' http://registry:8081/compatibility/subjects/topic-value/versions/latest
2. Nullable поля в Avro требуют union type: ["null", "string"] с "default": null
3. При добавлении нового поля всегда указывайте default: {"name": "new_field", "type": "string", "default": ""}
4. Используйте BACKWARD compatibility в Schema Registry для автоматической проверки

Связанные уроки:

Симптомы

• Чтение Avro-файла возвращает ошибку о negative length
• Файл частично читается, потом падает с AvroRuntimeException
• Data corruption после конкурентной записи

Причина

Avro Object Container File повреждён — sync marker потерян или данные в блоке усечены. Частая причина: writer не вызвал flush()/close(), S3 multipart upload завершился частично, или файл был перезаписан конкурентным процессом.

Решение

1. Проверьте целостность файла: avro-tools tojson --head 10 file.avro
2. Для частичного восстановления: читайте по блокам с try/catch, пропуская повреждённые блоки при обнаружении следующего sync marker
3. Предотвращение: всегда используйте context manager (with DataFileWriter(...) as writer:) для гарантии flush
4. Для Kafka: проверьте, что конфигурация producer включает acks=all и retries>0

Связанные уроки:

Симптомы

• Delta Lake отказывается записывать данные с новой схемой
• Streaming-запись в Delta-таблицу падает после изменения upstream-схемы
• Ошибка при добавлении новой колонки в существующую Delta-таблицу

Причина

Delta Lake по умолчанию не разрешает изменение схемы при записи. Новые колонки, изменённые типы или удалённые поля блокируются. Автоматическое mergeSchema включается явно.

Решение

1. Для добавления колонок: .option('mergeSchema', 'true').save() или spark.conf.set('spark.databricks.delta.schema.autoMerge.enabled', 'true')
2. В delta-rs: write_deltalake(dt, df, mode='append', schema_mode='merge')
3. Для overwrite со сменой схемы: .option('overwriteSchema', 'true').mode('overwrite').save()
4. Проверяйте совместимость до записи: dt.schema() vs df.schema

Связанные уроки:

Симптомы

• Два Spark-джоба одновременно пишут в одну Delta-таблицу — один падает
• Delta-запись завершается ошибкой конкурентного доступа
• OPTIMIZE и INSERT конфликтуют друг с другом

Причина

Delta Lake использует optimistic concurrency control — каждый коммит проверяет, что файлы не были изменены с момента начала транзакции. Конфликт возникает, когда два writer-а изменяют пересекающиеся partition-ы или файлы одновременно.

Решение

1. Используйте partition-based isolation: разные writer-ы пишут в разные партиции
2. Для append-only: конфликты маловероятны — два INSERT в разные файлы не конфликтуют
3. Добавьте retry-логику: for attempt in range(3): try: write(...) except ConcurrentAppendException: pass
4. Рассмотрите delta.isolationLevel=WriteSerializable (по умолчанию) vs Serializable для вашего use case

Связанные уроки:

Симптомы

• VACUUM команда отказывается удалять старые файлы
• Delta-таблица растёт без ограничений, занимая всё больше места
• VACUUM с retention < 7 дней блокируется

Причина

Delta Lake защищает от случайного удаления файлов, которые могут быть нужны для time travel или активных запросов. По умолчанию VACUUM удаляет только файлы старше 168 часов (7 дней). Retention < 7 дней требует явного override.

Решение

1. Для retention < 7 дней: spark.conf.set('spark.databricks.delta.retentionDurationCheck.enabled', 'false') ПЕРЕД вызовом VACUUM
2. Стандартный VACUUM: VACUUM table_name RETAIN 168 HOURS
3. В delta-rs: dt.vacuum(retention_hours=168, enforce_retention_duration=False)
4. Автоматизируйте через Airflow/cron: VACUUM → OPTIMIZE → VACUUM — в правильном порядке

Связанные уроки:

Симптомы

• Spark не может подключиться к Iceberg REST Catalog
• Ошибка при создании SparkSession с Iceberg-каталогом
• Таблицы Iceberg не видны через spark.sql

Причина

Конфигурация Spark-каталога Iceberg неполная или использует устаревший формат. Каждый тип каталога (REST, Hive, Glue, JDBC) требует свой набор параметров. REST Catalog требует uri, type, и правильный catalog-impl.

Решение

1. Для REST Catalog: spark.conf.set('spark.sql.catalog.my_catalog', 'org.apache.iceberg.spark.SparkCatalog') + spark.conf.set('spark.sql.catalog.my_catalog.type', 'rest') + spark.conf.set('spark.sql.catalog.my_catalog.uri', 'http://...')
2. Для Hive: type=hive + uri=thrift://... — не нужен catalog-impl
3. Для Glue: type=glue — подхватывает AWS credentials автоматически
4. Убедитесь, что iceberg-spark-runtime JAR добавлен в classpath: --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.0

Связанные уроки:

Симптомы

• ALTER TABLE ... ALTER COLUMN ... SET TYPE отклоняется Iceberg
• Попытка сменить nullable на required (или обратно) блокируется
• Изменение типа колонки не проходит валидацию

Причина

Iceberg разрешает только безопасные преобразования типов: int→long, float→double, расширение decimal precision. Изменение nullable → required запрещено, т.к. существующие файлы могут содержать NULL. Iceberg не перезаписывает данные при schema evolution.

Решение

1. Используйте только безопасные widening-преобразования: int→long, float→double, decimal(p1,s)→decimal(p2,s) где p2>p1
2. Для изменения nullability: создайте новую колонку с нужным типом, мигрируйте данные, удалите старую
3. Проверьте допустимые преобразования: документация Iceberg → Schema Evolution → Type promotion
4. Переименование безопасно — Iceberg использует Column ID, не имена

Связанные уроки:

Симптомы

• Два Spark-джоба пишут в Iceberg-таблицу — один получает CommitFailedException
• Iceberg-коммит не проходит из-за конфликта с другим writer
• INSERT OVERWRITE конфликтует с параллельным INSERT

Причина

Iceberg использует optimistic concurrency — коммит проверяет, что snapshot не изменился с момента начала операции. Конфликт возникает при конкурентных записях в пересекающиеся partition spec-ы. Retry-стратегия зависит от типа каталога.

Решение

1. Разнесите записи по partition-ам: разные writer-ы обновляют разные дни/часы
2. Настройте retry: write.commit.retry.num-retries=4 и write.commit.retry.min-wait-ms=100
3. Для append-only workloads конфликты минимальны — каждый writer создаёт свои файлы
4. Используйте Nessie или REST Catalog с branch isolation для изолированных writer-ов

Связанные уроки:

Симптомы

• DuckDB не может прочитать Parquet-файл, хотя PyArrow читает
• Ошибка footer при чтении файлов с S3
• Файлы нулевого размера (0 bytes) в выводе Spark

Причина

Parquet footer содержит схему и метаданные Row Groups и находится в конце файла. Если файл усечён (неполная запись, обрыв сети при загрузке на S3) или это пустой файл (0 строк, но footer есть), чтение падает. Spark иногда создаёт пустые файлы при записи пустых партиций.

Решение

1. Отфильтруйте пустые файлы: [f for f in files if os.path.getsize(f) > 0]
2. В DuckDB: SELECT * FROM parquet_scan('dir/*.parquet', union_by_name=true) — пропускает невалидные файлы
3. В Spark: .option('pathGlobFilter', '*.parquet').option('recursiveFileLookup', 'true')
4. Для S3: проверьте, что multipart upload завершён: aws s3api list-multipart-uploads

Связанные уроки:

Симптомы

• Hive/Spark не может прочитать ORC-файл — malformed ORC file
• Ошибка при чтении ORC после изменения compression codec
• ORC-файл, созданный одной версией Hive, не читается в другой

Причина

ORC footer повреждён или версия протокола несовместима. ORC magic bytes — ORC в начале файла. Причины: writer не закрыл файл, несовместимый compression codec (LZO требует отдельную библиотеку), или файл создан версией ORC, не поддерживаемой reader-ом.

Решение

1. Проверьте метаданные: orc-tools meta file.orc — покажет stripe-ы, типы, compression
2. Убедитесь, что compression codec доступен: LZO требует hadoop-lzo в classpath
3. Обновите ORC reader до версии, совместимой с writer: orc-tools data file.orc | head
4. Для восстановления: попробуйте orc-tools convert file.orc -o output.orc для пересоздания footer

Связанные уроки:

Симптомы

• Чтение Parquet-файла занимает неожиданно много времени
• Каждый файл содержит тысячи Row Groups по несколько КБ
• Memory consumption при чтении metadata выше ожидаемого

Причина

Слишком маленький row_group_size создаёт множество Row Groups в одном файле. Каждый Row Group хранит свою статистику — при тысячах мелких групп metadata footer раздувается и замедляет планирование. Типичная причина: запись по одной строке без буферизации.

Решение

1. Увеличьте row_group_size: pq.write_table(table, 'out.parquet', row_group_size=1_000_000)
2. В Spark: spark.conf.set('spark.sql.parquet.rowGroupSize', '134217728') — 128 MB по умолчанию
3. Перезапишите файл с оптимальным размером: прочитайте и запишите заново с правильным row_group_size
4. Для streaming: буферизуйте строки и записывайте batch-ами через ParquetWriter

Связанные уроки:

Симптомы

• Hive не находит вложенную колонку в ORC-файле
• Struct-поля в ORC читаются как NULL
• Переименование struct-поля ломает чтение

Причина

ORC привязывает вложенные колонки по позиции в schema (column ID), а не по имени. Если порядок полей в struct изменился между записью и чтением, данные читаются неправильно. В отличие от Iceberg (привязка по Column ID) и Avro (привязка по имени).

Решение

1. В Hive: SET hive.orc.schema.resolution=name вместо стандартного position-based resolution
2. Не переставляйте поля в struct при эволюции схемы — только добавляйте в конец
3. При рефакторинге: создайте новую таблицу с правильной схемой и INSERT INTO ... SELECT ...
4. Рассмотрите миграцию на Parquet или Iceberg для name-based schema evolution

Связанные уроки:

Симптомы

• pyiceberg не видит таблицу, которая существует в Spark SQL
• load_table() возвращает NoSuchTableError
• Разные каталоги видят разные наборы таблиц

Причина

pyiceberg подключён к другому каталогу, чем Spark. Частая ситуация: Spark использует Hive Metastore, а pyiceberg — REST Catalog. Каждый каталог имеет свой namespace таблиц. Также возможно несовпадение warehouse path.

Решение

1. Проверьте каталог в конфигурации: catalog.list_namespaces() и catalog.list_tables('db')
2. Убедитесь, что URI каталога совпадает: pyiceberg uri = Spark spark.sql.catalog..uri
3. Для Hive: pyiceberg требует thrift://host:9083, а не JDBC URL
4. Создайте таблицу через тот же каталог, через который планируете читать

Связанные уроки:

Симптомы

• PyArrow падает с OOM при чтении большого Parquet-файла
• Arrow allocation failed при конвертации pandas DataFrame
• Memory usage растёт неконтролируемо при конкатенации RecordBatch-ей

Причина

Arrow аллоцирует память через собственный MemoryPool. При чтении целого файла в память (read_table) все Row Groups загружаются одновременно. Конкатенация батчей (concat_tables) создаёт копию. String колонки с длинными значениями особенно затратны.

Решение

1. Читайте по батчам: pq.ParquetFile('file.parquet').iter_batches(batch_size=100_000)
2. Используйте column projection: pq.read_table('f.parquet', columns=['col1','col2'])
3. Для strings: используйте dictionary encoding: table.to_pandas(self_destruct=True) для освобождения Arrow memory при конвертации
4. Мониторьте пул: pa.total_allocated_bytes() для отслеживания текущего потребления

Связанные уроки:

Симптомы

• delta-rs / DuckDB не может прочитать Delta-таблицу, созданную Databricks
• Ошибка протокольной версии при чтении Delta
• Фичи (column mapping, deletion vectors) недоступны

Причина

Delta Lake использует версионирование протокола (reader/writer version). Новые фичи (deletion vectors, column mapping, v2 checkpoints) требуют более новый reader. Databricks часто создаёт таблицы с protocol v3/v7, а open-source клиенты поддерживают v1-2/v1-5.

Решение

1. Обновите delta-rs до последней версии: pip install deltalake>=0.17
2. В DuckDB: обновите до версии с поддержкой нужного протокола
3. Downgrade таблицы: ALTER TABLE t SET TBLPROPERTIES('delta.minReaderVersion'='1', 'delta.minWriterVersion'='2') — только если не используете v3+ фичи
4. Проверьте фичи: dt.protocol() — посмотрите, какие reader/writer features требуются

Связанные уроки:

Симптомы

• Protobuf-десериализация падает с invalid UTF-8 или wire type mismatch
• gRPC вызов возвращает INTERNAL error при парсинге response
• Данные, записанные одной версией proto, не десериализуются другой

Причина

Несовпадение proto-схемы: поле было переназначено на другой wire type (например, int32 → string на том же field number), или данные повреждены. Protobuf привязывает поля по номеру — изменение типа при сохранении номера нарушает десериализацию.

Решение

1. НИКОГДА не переиспользуйте номера полей — пометьте старые как reserved: reserved 5, 6; reserved "old_field";
2. Проверьте, что producer и consumer используют совместимые .proto: protoc --decode_raw < data.bin
3. Для диагностики wire type: protoc --decode_raw показывает raw field numbers и types
4. При эволюции: только добавляйте новые поля с новыми номерами, удаляйте — через reserved

Связанные уроки:

Симптомы

• Запросы к директории с тысячами мелких файлов работают медленно
• File listing на S3 занимает минуты
• Spark создаёт по одному файлу на каждый micro-batch

Причина

Каждый файл = overhead: metadata read, footer parsing, file open syscall, S3 GET request. При тысячах файлов overhead доминирует над полезным I/O. Streaming-записи и высокоселективные INSERT создают множество мелких файлов.

Решение

1. Delta Lake: OPTIMIZE table_name объединяет мелкие файлы в целевой размер (~1 GB)
2. Iceberg: spark.sql('CALL system.rewrite_data_files(table => "db.t", strategy => "binpack")')
3. Hudi: включите compaction для MoR или clustering для CoW
4. Spark: df.repartition(n).write.parquet(...) или .coalesce(n) для контроля числа файлов
5. Для streaming: увеличьте trigger interval или используйте foreachBatch с ручной буферизацией

Связанные уроки:

Симптомы

• Запись в Iceberg-таблицу после Partition Evolution завершается ошибкой
• Старые файлы не читаются после изменения партиционирования
• Непонятно, какая схема партиционирования используется

Причина

При Partition Evolution Iceberg создаёт новый partition spec, но старые файлы сохраняют прежний. Проблема возникает, если writer использует устаревший metadata — стартовал до ALTER TABLE, но коммитит после. Также: конфликт при concurrent ALTER TABLE + INSERT.

Решение

1. Выполните Partition Evolution отдельно от записи — сначала ALTER TABLE, потом INSERT
2. Проверьте текущий spec: table.spec() или SELECT * FROM my_table.partitions
3. Iceberg корректно читает данные с разными partition spec-ами — проблема только при записи
4. Используйте retry при конфликте: write.commit.retry.num-retries=4

Связанные уроки:

Симптомы

• Hudi-запись падает с конфликтом при конкурентном upsert
• Два Spark-джоба обновляют одну и ту же Hudi-таблицу
• MoR-таблица: compaction конфликтует с записью

Причина

Hudi использует file group locking для конкурентной записи. Два writer-а не могут обновлять один file group одновременно. В MoR-таблицах compaction и запись могут конфликтовать, если работают с пересекающимися file groups.

Решение

1. Включите multi-writer mode: hoodie.write.concurrency.mode=optimistic_concurrency_control + hoodie.write.lock.provider=org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider
2. Разнесите writer-ы по партициям: каждый джоб обновляет свой date/partition range
3. Для MoR: schedule compaction в периоды без записи или используйте inline compaction
4. Увеличьте retry: hoodie.cleaner.policy.failed.writes=LAZY для автоматической очистки failed writes

Связанные уроки:

Симптомы

• Регистрация новой версии Avro-схемы отклоняется Schema Registry
• Producer не может зарегистрировать схему с новым полем
• Ошибка 409 Conflict от Schema Registry API

Причина

Новая схема нарушает заданный уровень совместимости subject-а (BACKWARD, FORWARD, FULL). Типичные нарушения: удалено поле без default, добавлено required поле, изменён тип несовместимо.

Решение

1. Проверьте уровень совместимости: GET /config/{subject} — BACKWARD по умолчанию
2. Для BACKWARD: новые поля ОБЯЗАНЫ иметь default-значение
3. Проверьте совместимость до регистрации: POST /compatibility/subjects/{subject}/versions/latest
4. Для breaking changes: создайте новый subject (topic-v2-value) или временно смените compatibility на NONE (опасно!)

Связанные уроки:

Симптомы

• PyArrow/DuckDB не может прочитать Parquet-файл, созданный старым Spark или Hive
• Timestamp-колонка возвращает ошибку INT96
• Файлы из Hive warehouse не совместимы с pandas

Причина

Старые версии Spark (< 3.0) и Hive записывали timestamp как INT96 — нестандартный тип Parquet, занимающий 12 байт. Современные инструменты не поддерживают INT96 по умолчанию. Стандартный тип — INT64 timestamp с микросекундной точностью.

Решение

1. В PyArrow: pq.read_table('file.parquet', coerce_int96_timestamp_unit='ms') — конвертирует при чтении
2. В Spark при записи: spark.conf.set('spark.sql.parquet.outputTimestampType', 'TIMESTAMP_MICROS')
3. Перезапишите файлы: прочитайте с INT96-совместимым reader и запишите с стандартным timestamp
4. В DuckDB: SET arrow_lossless_conversion = true может помочь с некоторыми legacy файлами

Связанные уроки: