Хранение и алгоритмы сжатия данных - ПР 2 - Spark: ORC vs Parquet

PeaceDatabase/Controllers/DbAndDocumentsController.cs

@@ -97,7 +97,8 @@ public IActionResult DeleteDb([FromRoute] string db)
     public IActionResult AllDocs([FromRoute] string db, [FromQuery] int? skip, [FromQuery] int? limit, [FromQuery] bool? includeDeleted)
     {
         var docs = _svc.AllDocs(db, skip ?? 0, limit ?? 100, includeDeleted ?? true);
-        return Ok(new AllDocsResponse { Total = docs?.Count() ?? 0, Items = docs });
+        var list = docs?.ToList() ?? new List<Document>();
+        return Ok(new AllDocsResponse { Total = list.Count, Items = list });
     }
 
     // GET /v1/db/{db}/_seq
@@ -109,6 +110,15 @@ public IActionResult Seq([FromRoute] string db)
         return Ok(new SequenceResponse { Db = db, Seq = seq });
     }
 
+    // GET /v1/db/{db}/_stats
+    [HttpGet("{db}/_stats")]
+    [ProducesResponseType(typeof(StatsDto), StatusCodes.Status200OK)]
+    public IActionResult DbStats([FromRoute] string db)
+    {
+        var stats = _svc.Stats(db);
+        return Ok(stats);
+    }
+
     // ---- Поиск по полям ----
     public sealed class FindByFieldsRequest
     {

PeaceDatabase/Core/Services/IDocumentService.cs

@@ -46,6 +46,11 @@ public interface IDocumentService
         /// </summary>
         int Seq(string db);
 
+        /// <summary>
+        /// Статистика по базе (число документов, токены и т.д.).
+        /// </summary>
+        StatsDto Stats(string db);
+
         // ---------- Поиск ----------
         /// <summary>
         /// Поиск документов по равенствам полей и (опц.) числовому диапазону.

README.MD

@@ -34,7 +34,7 @@
 
 ### API Layer
 
-- **DbController** — базовые CRUD-операции, _all_docs, _changes.
+- **DbAndDocumentsController** — CRUD-операции, поиск, быстрая статистика по БД и другое.
 - **AttachmentsController** — работа с вложениями (PUT/GET/DELETE).
 - **ReplicationController** — интерфейс для репликации (push/pull).
 
@@ -62,7 +62,7 @@
 
 - **Core**: Document, IDocumentService.
 - **Storage**: InMemoryDocumentService (in-memory CRUD, AllDocs, Changes).
-- **API**: DbController (эндпоинты для CRUD и просмотра изменений).
+- **API**: DbAndDocumentsController (эндпоинты для CRUD и просмотра изменений).
 
 [![Architecture Diagram1](docs/images/Lab1.png)](docs/images/Lab1.png)
 

SaDCA-lab1.md

@@ -18,11 +18,11 @@
 
 ## Сравнение сериализации/десериализации документов в разных форматах (JSON, Binary, Protobuf)
 
-![Serialize](tools/docs/benchmarks_news/bytes_per_doc.png)
+![Serialize](tools/docs/SaDCA-lab1-custom-protobuf/benchmark_news/bytes_per_doc.png)
 
-![Deserialize](tools/docs/benchmarks_news/deserialize_vs_n.png)
+![Deserialize](tools/docs/SaDCA-lab1-custom-protobuf/benchmark_news/deserialize_vs_n.png)
 
-![BytesPerDoc](tools/docs/benchmarks_news/bytes_per_doc.png)
+![BytesPerDoc](tools/docs/SaDCA-lab1-custom-protobuf/benchmark_news/bytes_per_doc.png)
 
 
 ### Цели

SaDCA-lab2.md

@@ -0,0 +1,107 @@
+## SaDCA — Лабораторная 2. Spark DataSource V2 + Parquet/ORC
+
+### Цель
+Реализовать кастомный Source (DataSource V2) для Apache Spark на Java, читающий документы из PeaceDatabase по REST, построить простой ETL (read → transform → write) и сравнить эффективность хранения/чтения в Parquet и ORC.
+
+Справочные материалы:
+- Structured Streaming / Data Sources (Spark 3.5.x/4.x): [Structured Streaming Programming Guide](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html)
+- Hadoop winutils для Windows: [cdarlint/winutils](https://github.com/cdarlint/winutils)
+- [How to Install Apache Spark on Windows](https://phoenixnap.com/kb/install-spark-on-windows-10)
+
+### Окружение
+- ОС: Windows 11, JDK 11
+- Spark: 3.5.7 (Hadoop 3)
+- PeaceDatabase Web API: `http://localhost:5000`, БД: `news`
+
+### Архитектура решения
+- Кастомный коннектор (Java DataSource V2, batch read) → REST:
+  - Эндпоинт пагинации: `GET /v1/db/{db}/_all_docs?skip=&limit=&includeDeleted=`
+  - Схема набора:
+    - `id` (string), `rev` (string), `deleted` (boolean)
+    - `tags` (array<string>), `content` (string)
+    - `data_json` (string) — сырой JSON поля `data` (для совместимости)
+  - Планирование чтения: создаётся `InputPartition` на каждую страницу. При указании `maxRows` план основан на N (без доверия к `total` из API). Для случая без лимита выполняется «быстрый» пробег страниц до короткой/пустой.
+- Джоб (Spark App) делает преобразование и запись в две колоночные цели:
+  - Парсит `data_json` в столбцы: `link, headline, category, short_description, authors, date` (типизированные колонки улучшают компрессию/пушдаун)
+  - Примерные признаки: `tags_count = size(tags)`, `content_len = length(content)`
+  - Пишет в `out/parquet_#N` и `out/orc_#N`
+  - Измеряет размер папок, время записи и чтения (`count()`) и сохраняет в CSV
+
+### Spark DataSource / Job
+- Модуль в каталоге `spark/` (JDK 11, Spark 3.5.x)
+- Источник Spark V2 планирует чтение по `/v1/db/{db}/_stats` (без «пробега» страниц)
+- Поддерживаются таймауты и ретраи HTTP
+
+Полезные эндпоинты:
+- `GET /v1/db/{db}/_all_docs?skip=&limit=&includeDeleted=` — постраничный вывод документов
+- `GET /v1/db/{db}/_seq` — текущая последовательность изменений
+- `GET /v1/db/{db}/_stats` — статистика БД (docsTotal/docsAlive/docsDeleted)
+
+Ключевые файлы:
+- `spark/peacedb-datasource/src/main/java/io/peacedb/spark/PeacedbDataSource.java`
+- `spark/peacedb-datasource/src/main/java/io/peacedb/spark/read/*`
+- `spark/peacedb-job/src/main/java/io/peacedb/job/PeacedbToColumnar.java`
+- `tools/plot_peacedb_columnar.py`
+
+### Опции коннектора
+- `baseUrl` (по умолчанию `http://localhost:5000`)
+- `db` (по умолчанию `news`)
+- `outDir` - папка с результатами
+- `pageSize` (по умолчанию `100`)
+- `includeDeleted` (по умолчанию `false`)
+- `maxRows` — ограничение количества возвращаемых строк на стороне источника (используется в бенчмарке)
+- `connectTimeoutMs` / `readTimeoutMs` — таймауты HTTP
+- `retries`, `retryBackoffMs` - перезапуски
+
+### Сборка и запуск
+Установить Java, Spark, winutils
+
+Задать переменные окружения JAVA_HOME, HADOOP_HOME, SPARK_HOME
+
+```bash
+cd spark
+mvn -q -DskipTests package
+
+# Пример запуска (N=1/10/100/1000/10000/100000, warm‑up в одном прогоне)
+spark-submit --class io.peacedb.job.PeacedbToColumnar --master local[*] peacedb-job/target/peacedb-job-*.jar --baseUrl http://localhost:5000 --db news --outDir ./out --pageSize 200 --connectTimeoutMs 10000 --readTimeoutMs 60000 --retries 3 --retryBackoffMs 500
+
+# Построение графиков
+python tools/plot_peacedb_columnar.py spark/out/metrics.csv --out spark/out/plot.png
+```
+
+Примечание для Windows: при ошибке `NativeIO$Windows.access0` установить `winutils.exe` с dll для Hadoop 3.3.x и задать `HADOOP_HOME`/`PATH` (см. winutils выше). Для локальной ФС в коде/submit включены безопасные настройки (pure‑Java путь).
+
+
+### Методика измерений
+- Для каждого N в `[1, 10, 100, 1000, 10000, 100000]` читаем `maxRows=N`, считаем признаки, парсим `data_json` в колонки и:
+  1) Выполняем тёплый прогон (кэш + микро‑запись/чтение) для прогрева JIT/кодеков/коммит‑протокола
+  2) Измеряем запись и считывание Parquet и ORC (миллисекунды)
+- Результаты накапливаются в `out/metrics.csv` c колонками: `n,format,bytes,write_ms,read_ms,rows`
+
+### Результаты (пример, с тёплого прогона)
+
+![Metrics Plot](tools/docs/SaDCA-lab2-spark-orc-parquet/plot.png)
+
+[Таблица (CSV)](tools/docs/SaDCA-lab2-spark-orc-parquet/metrics.csv)
+
+Наблюдения:
+- Размеры у Parquet и ORC близки (после парсинга `data_json` в колонки). ORC слегка компактнее на этих данных.
+- Время записи: у Parquet и ORC соизмеримо на N до 10^4, на 10^5 ORC медленнее; начальные точки (1) могут отличаться из‑за недостаточного прогрева (cold start).
+- Время чтения: ORC немного быстрее, Parquet догоняет.
+
+Графики: `spark/out/plot.png` (лог‑ось X).
+
+### Решенные проблемы
+- Не хранить вложенные поля как «сырой JSON» — парсить в типизированные колонки (даёт компрессию и pushdown/column pruning).
+- Разогревать пайплайн перед замерами (JIT/кодеки/коммиттер), иначе первый пункт искажен.
+- На Windows учесть нюансы Hadoop Native/`winutils`.
+
+### Ограничения и улучшения
+- Сейчас данные можно загружать только "пакетами" (batch-only). В будущем можно добавить режим микробатчей или потоковой обработки (stream), используя поле `_seq` и watermark.
+- На данный момент фильтрация поддерживается только на стороне Spark. Можно улучшить, чтобы фильтры Spark автоматически переводились в запросы к API (`/v1/db/{db}/_find/*`), чтобы уменьшить объем передаваемых данных.
+- Структура данных (`data`) может меняться со временем. Чтобы не "ломать" обработку при изменении формата, стоит использовать функции вроде `from_json` с гибкой (нежесткой) схемой и автоматическим определением структуры (`schema_of_json` или автоинференс).
+
+### Выводы
+- Колоночные форматы выигрывают при типизации полей. Для новостного датасета состоящего преимущественно из текста, а не чисел ORC незначительно компактнее, времена записи/чтения для Parquet и ORC близки.
+
+

spark/.gitignore

@@ -0,0 +1,134 @@
+# JAVA
+
+# Compiled class file
+*.class
+
+# Log file
+*.log
+
+# BlueJ files
+*.ctxt
+
+# Mobile Tools for Java (J2ME)
+.mtj.tmp/
+
+# Package Files #
+*.jar
+*.war
+*.nar
+*.ear
+*.zip
+*.tar.gz
+*.rar
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+replay_pid*
+
+
+# SPARK
+
+#*#
+*.#*
+*.iml
+*.ipr
+*.iws
+*.pyc
+*.pyo
+*.swp
+*~
+.DS_Store
+.bsp/
+.cache
+.classpath
+.ensime
+.ensime_cache/
+.ensime_lucene
+.generated-mima*
+.idea/
+.idea_modules/
+.project
+.pydevproject
+.scala_dependencies
+.settings
+/lib/
+R-unit-tests.log
+R/unit-tests.out
+R/cran-check.out
+R/pkg/vignettes/sparkr-vignettes.html
+R/pkg/tests/fulltests/Rplots.pdf
+build/*.jar
+build/apache-maven*
+build/scala*
+build/zinc*
+cache
+checkpoint
+conf/*.cmd
+conf/*.conf
+conf/*.properties
+conf/*.sh
+conf/*.xml
+conf/java-opts
+conf/slaves
+dependency-reduced-pom.xml
+derby.log
+dev/create-release/*final
+dev/create-release/*txt
+dev/pr-deps/
+dist/
+docs/_site/
+docs/api
+sql/docs
+sql/site
+lib_managed/
+lint-r-report.log
+log/
+logs/
+out/
+project/boot/
+project/build/target/
+project/plugins/lib_managed/
+project/plugins/project/build.properties
+project/plugins/src_managed/
+project/plugins/target/
+python/lib/pyspark.zip
+python/.eggs/
+python/deps
+python/docs/_site/
+python/docs/source/reference/api/
+python/test_coverage/coverage_data
+python/test_coverage/htmlcov
+python/pyspark/python
+.mypy_cache/
+reports/
+scalastyle-on-compile.generated.xml
+scalastyle-output.xml
+scalastyle.txt
+spark-*-bin-*.tgz
+spark-tests.log
+src_managed/
+streaming-tests.log
+target/
+unit-tests.log
+work/
+docs/.jekyll-metadata
+docs/.jekyll-cache
+# For Hive
+TempStatsStore/
+metastore/
+metastore_db/
+sql/hive-thriftserver/test_warehouses
+warehouse/
+spark-warehouse/
+# For R session data
+.RData
+.RHistory
+.Rhistory
+*.Rproj
+*.Rproj.*
+.Rproj.user
+# For SBT
+.jvmopts
+
+
+out*/

spark/README.md

@@ -0,0 +1,47 @@
+## PeaceDB Spark DataSource (Java) and Job
+
+Build requires JDK 11+, Maven, Spark 3.5.1. PeaceDatabase Web API must run at `http://localhost:5000`.
+
+Data source features:
+- Plans partitions using fast `/v1/db/{db}/_stats` (no page probing for large datasets)
+- Configurable timeouts and automatic HTTP retries
+
+### Build
+
+```
+mvn -q -DskipTests package
+```
+
+Artifacts:
+- `peacedb-datasource/target/peacedb-datasource-*.jar`
+- `peacedb-job/target/peacedb-job-*.jar`
+
+### Run job
+
+```
+spark-submit \
+  --class io.peacedb.job.PeacedbToColumnar \
+  --master local[*] \
+  peacedb-job/target/peacedb-job-*.jar \
+  --baseUrl http://localhost:5000 \
+  --db news \
+  --outDir ./out \
+  --pageSize 200 \
+  --connectTimeoutMs 10000 \
+  --readTimeoutMs 60000 \
+  --retries 3 \
+  --retryBackoffMs 500
+```
+
+Outputs: `out/parquet_*`, `out/orc_*`, and `out/metrics.csv` with size and timing.
+
+Connector options (Spark.read.format("peacedb")):
+- `baseUrl` (default `http://localhost:5000`)
+- `db` (default `news`)
+- `pageSize` (default `100`)
+- `includeDeleted` (default `false`)
+- `maxRows` (for benchmarking)
+- `connectTimeoutMs` / `readTimeoutMs`
+- `retries` / `retryBackoffMs`
+
+