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

ЦЕЛИ

• Создать проект hybrid-json-datasource
• Разработать вспомогательные классы для работы с данными в следующих лабораторных

ЗАДАЧИ

org.apache.spark.sql.hybrid.JsonParser
Вам необходимо разработать класс для конвертации JSON строк в InternalRow package org.apache.spark.sql.hybrid import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.types.StructType class JsonParser(schema: StructType) { def toRow(input: Iterator[String]): Iterator[InternalRow] = { ??? } }

• schema - схема JSON документа
• input - итератор JSON документов. Каждая срока - это отдельный JSON документ вида {"foo": 0, "bar": "hello world"}

org.apache.spark.sql.hybrid.RowConverter
Вам необходимо разработать класс для конвертации InternalRow в JSON строки

package org.apache.spark.sql.hybrid import org.apache.spark.sql.catalyst.InternalRow class RowConverter(dataType: StructType) { def toJsonString(input: Iterator[InternalRow]): Iterator[String] = { ??? } }

• schema - схема InternalRow
• input - итератор InternalRow. На выходе должен получится Iterator[String], содержащий JSON строки вида {"foo": 0, "bar": "hello world"}. Каждая InternalRow превращается в одну JSON строку

ЦЕЛИ

• Разработать функционал для чтения и записи формата hybrid-json

ЗАДАЧИ

Использование коннектора
// writing
spark .range(0, 10, 1, 1) .write.format("hybrid-json") .option("path", "/tmp/foo.json") .option("objectName", "test01") .save()

// reading val df = spark.read.format("hybrid-json").option("objectName", "test01").load()

Параметры:

• MONGO_URI - переменная окружения, в которой задается адрес для подключения к БД Mongo. Пример: mongodb://localhost:27017
• path - локальный путь к директории, в которую необходимо произвести запись заданного DataFrame (используется только при записи)
• objectName - имя объекта (для аналогии можно считать, что это имя некоторой таблицы, в которую будет делаться запись)

ЦЕЛИ

• Добавить механизм predicate pushdown в hybrid-json

ЗАДАЧИ

Вам необходимо:

• реализовать механизм индексирования, аналогичному в parquet, при использовании которого для каждого файла для каждой колонки необходимо будет посчитать ее минимальное и максимальное значение и сохранить это в index_store.file_index в поле columnStats в следующем формате:

> use index_store switched to db index_store > db.file_index.findOne() { "_id" : ObjectId("619f54f0f93d4b9795eb4606"), "objectName" : "test01", "path" : "/tmp/foo.json/ca8220e2-bd77-4732-a837-9071854086a0.json", "commitMillis" : NumberLong("1637831920142"), "columnStats" : [ { "name" : "id2", "min" : 0, "max" : 38 }, { "name" : "id", "min" : 0, "max" : 19 } ] }

• cтруктура поля columnStats - это массив словарей, в каждом словаре name - имя колонки, min - ее минимальное значение в файле, max - ее максимальное значение в файле
• запись columnStats должна происходить после успешной записи файла, вместе с остальными полями документа в index_store.file_index
• для упрощения задачи необходимо реализовать поддержку columnStats только для полей, имеющих тип IntegerType
• реализовать механизм predicate pushdown, при котором будут читаться только те файлы, которые пройдут по columnnStats. Должны поддерживаться следующие фильтры
• EqualTo ( == в DataFrame API)
• GreaterThan ( > в DataFrame API)
• GreaterThanOrEqual ( >= в DataFrame API)
• LessThan ( < в DataFrame API)
• LessThanOrEqual ( <= в DataFrame API)

ЦЕЛИ

• Добавить поддержку Structured Streaming API в hybrid-json

ЗАДАЧИ

Добавление поддержки StreamSourceProvider
val df = spark.readStream.format("hybrid-json").option("objectName", "test01").load()

• На каждом батче необходимо вычитывать все файлы, начиная с последнего оффсета
• В качестве оффсета необходимо использовать максимальное значение commitMillis, прочитанное из index_store.file_index из последнего батча

Добавление поддержки StreamSinkProvider
val sq = df.writeStream .format("hybrid-json") .option("objectName","test01") .option("path","/tmp/test01.json") .option("checkpointLocation", "/tmp/chk01.chk") .start() sq.awaitTermination()

• Каждый батч должен записывать по аналогии с обычным .write