Рамки для представления обработки данных в качестве конвейера

Большая часть обработки данных может быть предусмотрена как конвейер компонентов, выход одного из которых подается на вход другого. Типичный конвейер обработки:

reader | handler | writer

Как фольга для начала этого обсуждения, рассмотрим объектно-ориентированную реализацию этого конвейера, где каждый сегмент является объектом. Объект handler содержит ссылки на объекты reader и writer и имеет метод run, который выглядит так:

define handler.run:
  while (reader.has_next) {
    data = reader.next
    output = ...some function of data...
    writer.put(output)
  }

Схематически зависимости:

reader <- handler -> writer

Теперь предположим, что я хочу вставить новый сегмент конвейера между считывателем и обработчиком:

reader | tweaker | handler | writer

Опять же, в этой реализации OO tweaker будет оберткой вокруг объекта reader, а методы tweaker могут выглядеть примерно как (в некотором псевдо-императивном коде):

define tweaker.has_next:
  return reader.has_next

define tweaker.next:
  value = reader.next
  result = ...some function of value...
  return result

Я нахожу, что это не очень сложная абстракция. Некоторые проблемы:

tweaker может использоваться только с левой стороны handler, то есть я не могу использовать приведенную выше реализацию tweaker для формирования этого конвейера:

читатель | обработчик | твикер | автор
Я хотел бы использовать ассоциативное свойство конвейеров, чтобы этот конвейер:

читатель | обработчик | автор

может быть выражена как:

reader | p

где p - конвейер handler | writer. В этой реализации OO я должен частично создать экземпляр объекта handler

В некоторой степени из повторения (1), объекты должны знать, если они "толкают" или "тянут" данные.

Я ищу фреймворк (не обязательно OO) для создания конвейеров обработки данных, которые решают эти проблемы.

Я отметил это с помощью Haskell и functional programming, потому что я считаю, что функциональные концепции программирования могут быть полезны здесь.

В качестве цели было бы неплохо создать такой конвейер:

                     handler1
                   /          \
reader | partition              writer
                   \          /
                     handler2

В какой-то перспективе оболочки оболочки Unix решают многие из этих проблем с помощью следующих решений:

Компоненты трубопровода выполняются асинхронно в отдельных процессах
Объекты труб опосредуют передачу данных между "толкателями" и "съемниками"; то есть они блокируют авторов, которые слишком быстро записывают данные и читают, которые слишком быстро читают.
Вы используете специальные коннекторы < и > для подключения пассивных компонентов (например, файлов) к конвейеру

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

Спасибо!

Ответ 1

Да, arrows почти наверняка ваш человек.

Я подозреваю, что вы довольно новичок в Haskell, просто основываясь на тех вещах, которые вы говорите в своем вопросе. Стрелки, вероятно, кажутся довольно абстрактными, особенно если то, что вы ищете, является "каркасом". Я знаю, мне потребовалось некоторое время, чтобы действительно понять, что происходит со стрелками.

Таким образом, вы можете посмотреть на эту страницу и сказать "да, это похоже на то, что я хочу", а затем немного потерять себя, как начать использовать стрелки для решения проблемы. Итак, вот немного руководства, чтобы вы знали, что вы смотрите.

Стрелки не решают вашу проблему. Вместо этого они дают вам язык, на котором вы можете использовать свою фразу. Вы можете обнаружить, что какая-то предопределенная стрелка выполнит задание - возможно, что-то вроде kleisli arrow, но в конце дня вам захочется реализовать стрелку (предопределенные просто дают вам простые способы их реализации), которая выражает что вы подразумеваете под "процессором данных". В качестве почти тривиального примера предположим, что вы хотите реализовать свои процессоры данных с помощью простых функций. Вы должны написать:

newtype Proc a b = Proc { unProc :: a -> b }

-- I believe Arrow has recently become a subclass of Category, so assuming that.

instance Category Proc where
    id = Proc (\x -> x)
    Proc f . Proc g = Proc (\x -> f (g x))

instance Arrow Proc where
    arr f = Proc f
    first (Proc f) = Proc (\(x,y) -> (f x, y))

Это дает вам механизм для использования различных комбинаторов стрелок (***), (&&&), (>>>) и т.д., а также обозначение стрелки, которое довольно хорошо, если вы делаете сложные вещи. Итак, как указывает Даниэль Фишер в комментарии, трубопровод, который вы описали в своем вопросе, может состоять из:

reader >>> partition >>> (handler1 *** handler2) >>> writer

Но здорово, что вам решать, что вы подразумеваете под процессором. Можно реализовать то, что вы упомянули о каждом процессоре, который разветвляет поток аналогичным образом, используя другой тип процессора:

newtype Proc' a b = Proc (Source a -> Sink b -> IO ())

И затем соответствующим образом реализуем комбинаторы.

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

Одним из моих первых нетривиальных проектов Haskell было внедрение стрелки для квантовой запутанности; этот проект был тем, что заставило меня действительно начать понимать мышление Хаскелла, что стало поворотным моментом в моей карьере программирования. Возможно, этот ваш проект сделает то же самое для вас?: -)

Ответ 2

Благодаря ленивой оценке, мы можем выразить конвейеры в терминах обычных функциональных композиций в Haskell. Вот пример, который вычисляет максимальную длину строки в файле:

main = interact (show . maximum . map length . lines)

Все здесь - обычная функция, например,

lines :: String -> [String]

но благодаря ленивой оценке эти функции обрабатывают только инкрементный процесс и только столько, сколько необходимо, так же, как и в UNIX-канале.

Ответ 3

пакет перечислителя для Haskell - хорошая основа для этого. Он определяет три типа объектов:

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

Эти три типа объектов объединяются в конвейер обработки потока, и вы даже можете иметь несколько счетчиков и итерации в одном конвейере (когда один закончен, следующий занимает свое место). Сложно написать один из этих объектов с нуля, но есть много комбинаторов, которые могут использоваться для превращения регулярных функций в процессоры потоков данных. Например, этот конвейер считывает все символы из stdin, преобразует их в верхний регистр с помощью функции toUpper, а затем записывает их в stdout:

ET.enumHandle stdin $$ ET.map toUpper =$ ET.iterHandle stdout

где модуль Data.Enumerator.Text был импортирован как ET.

Ответ 4

Структура Yesod использует библиотеку каналов Haskell в виде conduit.