Функциональное ядро в виде конвейера на Python

Kate

Administrator
Команда форума
Главная задача этого поста – показать один мало применяемый в Python подход к работе функциями в рамках более общей функциональной парадигмы, в которой функциями можно манипулировать точно так же, как и любыми другими объектами: присваивать переменным, передавать в качестве аргументов в другие функции, возвращать из функций и включать в последовательности в качестве их элементов. Тем, кому требуется освежить свою память о функциональном программировании на языке Python, рекомендую перейти по ссылке к моему посту об основах ФП на Python.

Конвейер обработки данных
Конвейер обработки данных
Функциональный стиль программирования очень близок к тому, как размышляет человек во время решения задачи. «Пусть дано x. В целях решения задачи с этими данными необходимо выполнить серию преобразований. Сначала применить к ним f и получить результирующие данные x'. Затем к новым данным применить f2 и получить новые результирующие данные x''и т.д.

Как оказалось, такой образ мыслей отлично укладывается в то, что называется конвейером обработки данных. Конвейер обработки данных состоит из связанных между собой узлов, т.е. функций. Узел характеризуется набором входных и выходных каналов, по которым могут передаваться объекты. Узел ожидает появления определенного набора объектов на своем входном канале, после чего проводит вычисления и порождает объект(ы) на своем выходном канале, которые передаются в следующий узел в конвейере.

Конвейер в том виде, который он тут предлагается, позволяет (1) четче мониторить входы и выходы каждого шага внутри конвейера (2) легко отлаживать любой шаг и соответственно отлавливать дефекты, вставляя шаг debug, (3) но главное концентрировать всю логику в одном месте, которую легко можно улавливать одним взглядом.

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

2
|> ( fun x -> x + 5)
|> ( fun x -> x * x)
|> ( fun x -> x.ToString() )
Здесь входные данные, в данном случае число 2, последовательно обрабатываются серией лямбда-функций. Аналогичный конвейер можно имплементировать на языке Python, но для этого нужно написать специальную функцию, и, разумеется, это будет функция более высокого порядка:

# Конвейер обработки данных
def pipe(data, *fseq):
for fn in fseq:
data = fn(data)
return data
Приведенный ниже пример демонстрирует работу конвейера на языке Python:

pipe(2,
lambda x: x + 5,
lambda x: x * x,
lambda x: str(x))
или в более удобном виде:

add = lambda x: lambda y: x + y
square = lambda x: x * x
tostring = lambda x: str(x)

pipe(2,
add(5),
square,
tostring)
Число 2 проходит серию преобразований, и в результате будет получено строковое значение '49'. По сравнению с функцией reduce, в которой переданная в качестве аргумента одна единственная редуцирующая функция по очереди применяется к последовательности данных, в функции pipe наоборот последовательность функций применяется к обновляемым данным.

Функция pipeполучает два аргумента: входные данные dataи последовательность функций fseq. Во время первой итерации цикла forданные передаются в первую функцию из последовательности. Эта функция обрабатывает данные и возвращает результат, замещая переменную data новыми данными. Затем эти новые данные отправляются во вторую функцию и т.д. до тех пор, пока не будут выполнены все функции последовательности. По завершению своей работы функция pipe возвращает итоговые данные. Это и есть конвейер обработки данных.

Примечание. В приведенном выше примере функции pipe использован оператор упаковки *. В зависимости от контекста оператор * служит для упаковки получаемых нескольких аргументов в одну параметрическую переменную либо распаковки списка передаваемых в функцию аргументов.

Когда он используется в параметре функции, как в приведенном выше примере, он служит для упаковки всех аргументов в одну параметрическую переменную. Например,

def my_sum(*args): # Упаковка в список
return sum(args)

my_sum(1, 2, 3, 4, 5)
Когда он используется при вызове функции он служит для разложения передаваемого списка на отдельные аргументы. Например,

def fun(a, b, c, d):
print(a, b, c, d)

my_list = [1, 2, 3, 4]
fun(*my_list) # Разложение на четыре аргумента
Справедливости ради следует привести имплементацию конвейера обработки данных в парадигме объектно-ориентированного программирования (программный код позаимствован отсюда):

class Factory:
def process(self, input):
raise NotImplementedError

class Extract(Factory):
def process(self, input):
print("Идет извлечение...")
output = {}
return output

class Parse(Factory):
def process(self, input):
print("Идет разбор...")
output = {}
return output

class Load(Factory):
def process(self, input):
print("Идет загрузка...")
output = {}
return output

pipe = {
"Извлечь" : Extract(),
"Разобрать" : Parse(),
"Загрузить" : Load(),
}

inputs = {}
# Конвейерная обработка
for name, instance in pipe.items():
inputs = instance.process(inputs)
Вывод программы:

Идет извлечение...
Идет разбор...
Идет загрузка...
Здесь в цикле for результат на выходе из предыдущего шага подается на вход следующего шага, как того и требует стандартный конвейер. Однако проблема с объектно-ориентированным подходом заключается в том, что он неявно привносит всю свою ОО среду. «Вы хотели получить банан, но получили банан в месте с гориллой, которая его держит, и все джунгли в придачу» - эта цитата принадлежит Джо Армстронгу, создателю функционально-ориентированного языка Erlang.

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

Функциональное вычисление факториала числа​

В приведенном ниже примере показана нерекурсивная версия алгоритма вычисления факториала (factorial) и его рекурсивной версия на основе более эффективной хвостовой рекурсии (factorial_rec). Детали имплементации обеих функций в данном случае не важны. Они приводятся в качестве примеров, на которых будет продемонстрирована работа конвейера обработки данных. Результат выполнения программы показан ниже.

Следует сразу оговориться, что эта версия является предварительной, и ее чтение - занятие не из приятных ;-), т.к. в ней вся логика развернута внутри конвейера.

# Эта программа демонстрирует
# функциональную версию вычисления факториала

def main():
# Конвейер (ядро c нерекурсивным алгоритмом факториала)
pipe(int(input('Введите неотрицательное целое число: ')),
lambda n: (n, reduce(lambda x, y: x * y, range(1, n + 1))),
lambda tup:
print(f'Факториал числа {tup[0]} равняется {tup[1]}'))

# Вызвать главную функцию
main()
Вывод программы:

Введите неотрицательное целое число: 4 (Enter)
Факториал числа 4 равняется 24
Вторая лямбда-функция в последнем узле конвейера получает кортеж, состоящий из введенного пользователем числа и полученного результата.

Приведенная ниже расширенная версия программы вычисления факториала имеет более читабельный вид, т.к. шаги конвейера - алгоритмы - выделены в отдельные функции. В нее также добавлен функционал валидации входных данных. Чуть позже будет дано пояснение.

# Эта программа демонстрирует
# функциональную версию функции вычисления факториала

def get_int(msg=''):
return int(input(msg))

def main():
# Алгоритм 1. Рекурсивная версия с хвостовой рекурсией
def factorial_rec(n):
fn = lambda n, acc=1: acc if n == 0 else fn(n - 1, acc * n)
return n, fn(n)

# Алгоритм 2. Нерекурсивная версия
def factorial(n):
return n, reduce(lambda x, y: x * y, range(1, n + 1))

# Ввод данных
def indata():
def validate(n): # Валидация входных данных
if not isinstance(n, int):
raise TypeError("Число должно быть целым.")
if not n >= 0:
raise ValueError("Число должно быть >= 0.")
return n
msg = 'Введите неотрицательное целое число: '
return pipe(get_int(msg), validate)

# Вывод данных
def outdata():
def fn(data):
n, fact = data
print(f'Факториал числа {n} равняется {fact}')
return fn

# Конвейер (функциональное ядро)
pipe(indata(), # вход: - выход: int
factorial, # вход: int выход: кортеж
outdata()) # вход: кортеж выход: -

# Вызвать главную функцию
main()
Вывод программы:

Введите неотрицательное целое число: 4 (Enter)
Факториал числа 4 равняется 24
Функциональным ядром приведенной выше программы являются строки:

pipe(indata(),
factorial,
outdata())
Они представлены конвейером из трех узлов, т.е. функциями indata, factorialи outdata. Функция indataзанимается получением данных от пользователя, которые затем передаются по конвейеру дальше. Функция factorialявляется собственно обрабатывающим алгоритмом, в данном случае нерекурсивной функцией вычисления факториала, которая получает данные, их обрабатывает и передает по конвейеру дальше. И функция outdataполучает данные и показывает их пользователю. Обратите внимание, что функция indata имеет свой собственный конвейер, который состоит из получения данных от пользователя и их валидации.

Следует отметить два важных момента. Во-первых, передаваемые от узла к узлу данные должны соответствовать какому-то определенному протоколу. Во-вторых, количество узлов может быть любым.

Такая организация программного кода:

  • Позволяет менять узлы конвейера на другие с целью тестирования различных и более эффективных имплементаций алгоритмов. Например, вместо нерекурсивной функции factorial, можно поместить рекурсивную функцию factorial_rec.
pipe(indata(), factorial_rec, outdata())
  • Облегчает проведение отладки программы, позволяя на каждом стыке вставлять отладочный код с целью проверки промежуточных результатов и тестирования производительности отдельных узлов.
Например, рассмотрим вторую возможность – отладку. В этом случае можно написать вспомогательную функцию debug:

def debug(data):
print(data)
return data
И затем ее вставить в конвейер, чтобы проверить результаты работы отдельных узлов конвейера:

pipe(indata(), debug, factorial, debug, outdata())
Если выполнить программу в таком варианте, то будут получены следующие результаты:

Вывод программы:

Введите неотрицательное целое число: 4 (Enter)
4
(4, 24)
Факториал числа 4 равняется 24

Как видно из результатов, на вход в функцию factorial поступает введенное пользователем значение 4, а на выходе из нее возвращается кортеж с исходным числом и полученным результатом (4, 24). Этот результат показывает, что программа работает, как и ожидалось. Как вариант, вместо проверочной функции debugможно написать функцию-таймер, которая могла бы хронометрировать отдельные узлы конвейера.

Приведем еще пару примеров с аналогичной организацией программного кода на основе функционального ядра в виде конвейера.

Функциональное вычисление последовательности Фибоначчи​

# Эта программа демонстрирует
# функциональную версию вычисления последовательности Фибоначчи

def main():
# Алгоритм
def fibonacci(n, x=0, y=1):
# Функция fib возвращает n-ое число последовательности.
fib = lambda n, x=0, y=1: x if n <= 0 else fib(n - 1, y, x + y)
# Функция reduce собирает результаты в список acc
acc = []
reduce(lambda _, y: acc.append(fib(y)), range(n + 1))
return n, acc

# Валидация входных данных
def validate(n):
if not isinstance(n, int):
raise TypeError("Число должно быть целым.")
if not n >= 0:
raise ValueError("Число должно быть ноль положительным.")
if n > 10:
raise ValueError("Число должно быть не больше 10.")
return n

# Ввод данных
def indata():
msg = 'Введите неотрицательное целое число не больше 10: '
return pipe(get_int(msg), validate)

# Вывод данных
def outdata():
def fn(data):
n, seq = data
msg = f'Первые {n} чисел последовательности Фибоначчи:'
print(msg)
[print(el) for el in seq]
return fn

# Конвейер (функциональное ядро)
pipe(indata(), fibonacci, outdata())

# Вызвать главную функцию.
main()

Вывод программы
Введите неотрицательное целое число не больше 10: 10 (Enter)
Первые 10 чисел последовательности Фибоначчи:
1
1
2
3
5
8
13
21
34
55

Функциональное суммирование диапазона значений последовательности​

# Эта программа демонстрирует
# функциональную версию суммирование
# диапазона значений последовательности

def main():
# Алгоритм
def range_sum(data):
seq, params = data
fn = lambda start, end: 0 if start > end \
else seq[start] + fn(start + 1, end)
return fn(*params)

# Ввод данных
def indata():
seq = [1, 2, 3, 4, 5, 6, 7, 8, 9]
params = (2,5) # params - это параметры start, end
return seq, params

# Вывод данных
def outdata():
def f(data):
msg = 'Сумма значений со 2 по 5 позиции равняется '
print(msg, format(data), sep='')
return f

# Конвейер (функциональное ядро)
pipe(indata(), range_sum, outdata())

# Вызвать главную функцию.
main()

Вывод программы
Сумма значений со 2 по 5 позиции равняется 18
Приведенный в настоящей главе материал имеет ознакомительный характер и служит для иллюстрации возможностей функциональной парадигмы программирования на Python с целью дальнейших самостоятельных исследований и чтобы побудить программистов дать функциональному стилю шанс. Исходный код поста находится в моем репо на Github. Материал поста использовался в качестве авторского дополнения в русском переводе книги Strating Out with Python. В англоязычном интернете можно найти материал, в котором рассматриваются шаблоны ветвления потока данных внутри конвейера. Например, несколько шаблонов дизайна конвейеров можно найти здесь, здесь и здесь. Очень рекомендую презентацию на Youtube «Конвейеризация на Python - конвейеры в приложениях науки о данных».

Вместе с тем, следует помнить о том, что язык Python скорее служит хорошим введением в ФП, если брать его функционал map/filter/reduce/zip и functools. Отсутствие оптимизации стека для хвостовой рекурсии в пределе делает невозможным полное и чистое ФП. Вместе с тем внедрение его принципов помогает четче излагать логику, избегать громоздкости ООП и смотреть на задачу под иным функциональным углом зрения, который ближе тому, как мы думаем.


Источник статьи: https://habr.com/ru/post/555370/
 
Сверху