Потоковая обработка данных в go

При использовании языка Go чувствуется нехватка хорошей библиотеки для работы с потоками данных. Имеющиеся библиотеки в основном написаны во времена до дженериков и имеют не очень удобный интерфейс.

С выходом Go 1.18 появилась возможность реализовать библиотеку потоковой обработки данных goio, вдохновлённую превосходными Scala-библиотеками cats-effect и fs2.

Потребности потоковых программ

При обработке данных, объём которых превышает объём памяти вычислительного узла, необходимо использовать технологии потоковой обработки данных. При этом данные загружаются в память по частям, выполняется их обработка, а затем данные выгружаются из памяти.

Для написания таких программ необходимо каким-либо образом решить ряд вопросов:

• обработка ошибок;

• представление конечных автоматов, реализующих отдельные этапы обработки данных;

• композиция конечных автоматов;

• использование нескольких ядер для параллельной обработки данных.

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

На мой вгляд, наиболее интересное решение предлагают Scala-библиотеки cats-effect и fs2.

Обработка ошибок

Многие функции не являются «тотальными». То есть они не всегда возвращают осмысленное значение. В языке Go принято представлять такие частичные функции с помощью возврата пары — значения функции и ошибки:

func f[A any]() (A, error) {   ... }

При этом вызывающая сторона должна сразу проверить наличие ошибки:

a, err = f() if err != nil {   return }

Такой способ представления ошибок создаёт неудобства при композиции нескольких вычислений.

Принятым в функциональном программировании способом представления вычислений, которые могут завершиться неудачей, является IO[A]. IO[A] — объект, представляющий вычисление, которое в случае успеха возвращает значение типа A. (Это, несколько упрощённое представление IO, позволяет сфокусироваться на аспекте, связанном с обработкой ошибок.)

В рассматриваемой библиотеке goio этот тип реализован. Например, вышеприведённая функция f может быть непосредственно «завёрнута» в IO[A]:

fio := io.Eval(f)

(Также имеется целый набор удобных способов заворачивания вычислений в IO[A] — Lift, Fail, Delay.)

Чтобы выполнить вычисление IO[A], используется метод UnsafeRunSync:

a, err = io.UnsafeRunSync(fio)

операции создания IO и выполнения почти эквивалентны непосредственному вызову f. Отличие заключается в том, что при возникновении panic , производится восстановление (recover) в error.

Какие возможности предоставляет тип IO[A]?

Главная возможность — композиция вычислений. Выполнить одно вычисление, результат подставить в другое, и всё вместе представить как одно вычисление. Такая операция в библиотеке goio носит название FlatMap (также известна как bind):

func g[A any, B any](a A) io.IO[B] { ... }  h := io.FlatMap(f, g)

Предусмотрены и другие привычные комбинаторы — Map, AndThen, Sequence…

Потоковая обработка данных

Если IO[A] представляет собой однократное вычисление,то Stream[A] представляет потенциально бесконечный поток элементов A. Выполняя некоторое действие, можно последовательно получить все элементы из потока.

Создать поток можно, например, из массива данных:

ints := stream.LiftMany(1, 2, 3)

Или можно создать бесконечный поток с использованием генератора:

var nats = stream.Unfold(0, func(s int) int { return s + 1 })

Также имеются и другие способы создания потоков — Empty, Lift, Repeat, Eval.

Обработка потока осуществляется конечным автоматом, имеющим вход и выход. Простейший случай — вход преобразуется в выход без внутреннего состояний. Такое преобразование осуществляется комбинатором Map:

powers := stream.Map(nats, func (i int) int { return i * i })

Универсальный механизм, позволяющий представить произвольный конечный автомат с состоянием — StateFlatMapWithFinish (см. пример в тестах). Имеются и другие комбинаторы — Filter, Flatten, AndThen, Take, Drop,…

Чтобы выполнить все вычисления и получить результат обработки потока, используются финальные операции — DrainAll, ToSlice, Collect,…

В библиотеке также реализованы простые средства для работы с текстовыми файлами или http-запросами — ReadLines, WriteLines.

Параллельное выполнение

Асинхронная операция выполняется вне текущего потока выполнения. То есть её можно запустить, передав в качестве аргумента callback, и продолжить работу текущего потока:

//io.Async[A any](k func(Callback[A])) IO[A]  helloIO := io.Async(func (cb Callback[string]) {   // start operation   go func(){     time.Sleep(100 * time.Millisecond)     // at the end call cb:     cb("hello", nil)   }() })

Когда асинхронная операция завершится, она должна вызвать переданный callback. (Если callback не вызывается, то получится операция Never.)

Текущий поток, после того, как выполнит свою оставшуюся работу, в какой-то момент может захотеть дождаться (Join) результата асинхронной операции. При этом текущий поток блокируется до получения результата (или ошибки).

Для представления такой семантики параллельного исполнения используется Fiber[A] — вычисление, запускаемое в отдельной горутине, результат которой мы можем получить с помощью Join.

Для создания Fiber используется функция Start(IO[A]) Fiber[A].

sleep100msAndReturnHello := io.Start(io.Pure(func() string {   time.Sleep(100 * time.Millisecond)   return "hello" })) dosomething := ... // выполняем какие-то действия после запуска. getHello := io.FlatMap(sleep100msAndReturnHello, func (fiber Fiber[A]) IO[string] {   return io.FlatMap(dosomething, fiber.Join()) }) hello, err := io.UnsafeRunSync(getHello)

Для создания сразу большого числа параллельных вычислений используются комбинаторы Parallel и ConcurrentlyFirst.

Ресурсы

Одним из аспектов корректной обработки ошибок является освобождение ресурсов по окончании операций независимо от того, произошла ошибка или нет. Для представления таких ресурсов используется тип Resource[A]. Ресурсами могут быть соединения с базой данных, файлы, каналы… Основная операция, предусмотренная для ресурса — Use, позволяющая в ограниченном конексте получить доступ к экземпляру ресурса. При выходе за границы контекста производится автоматическое закрытие ресурса. Пример работы с ресурсами.

Каналы

Для организации межпоточного взаимодействия в языке Go предусмотрены каналы. Для корректной работы с каналами в библиотеке goio предусмотренна поддержка на всех уровнях.

В пакете io : открытие/закрытие канала, отправка и чтение данных.

В пакете stream: отправка потока данных в канал, формирование потока данных из канала, использование внешней системы, подключенной посредством входного и выходного каналов.

В пакете resource: представление канала в виде ресурса — UnbufferedChannel[A any]() Resource[chan A].

Функциональные операции с массивами (slice)

Без привычных операций Map, FlatMap, Filter, … работа с массивами не очень удобна. В библиотеке goio содержатся эти операции, реализованные с использованием generics.

Заключение

Библиотека goio содержит удобный набор инструментов для написания надёжных и корректных программ, осуществляющих потоковую обработку данных. Реализованы:

• представление вычислений, которые могут возвращать ошибки, и комбинация таких вычислений;

• представление абстрактных потоков данных;

• представление конечных автоматов, реализующих отдельные этапы обработки данных;

• композиция цепочки конечных автоматов, выполняющих операции над потоками данных;

• параллельная обработка данных;

• взаимодействие с другими частями программ, реализованными с использованием каналов Go.

Библиотека доступна по лицензии MIT. PR приветствуются.