Как стать автором
Обновить

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

Время на прочтение 12 мин
Количество просмотров 8.8K

Главная задача этого поста – показать один мало применяемый на языке Python архитектурный шаблон под названием «функциональное ядро - императивная оболочка», в котором функциональный код концентрируется внутри, а императивный код выносится наружу в попытке свести на нет недостатки каждого из них. Известно, что функциональные языки слабы при взаимодействии с «реальным миром», в частности с вводом данных пользователем, взаимодействием с графическим интерфейсом или другими операциями ввода-вывода. В рамках такого подхода весь код, связанный с вводом-выводом, выталкивается наружу, и внутри остается только функционально-ориентированный код.

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

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

Конвейер обработки данных
Конвейер обработки данных

В отличие от объектно-ориентированного стиля мышления, в котором все внимание сосредоточено на том, как нужно делать, чтобы решить задачу, функциональный стиль мышления сосредоточен на том, что нужно делать. У нас есть данные x. И требуется получить из них некий результат. С этой целью сначала надо применить к ним преобразование f , затем к полученным данным x'применить f2, затем к этим новым даннымx''применить следующее преобразование f3 и т.д. вплоть до получения нужного результата.

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

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

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

 2                            
 |> ( fun x -> x + 5)         
 |> ( fun x -> x * x)         
 |> ( fun x -> x.ToString() ) 

Здесь входные данные, в данном случае число 2, последовательно обрабатываются серией лямбда-функциями, и в результате будет получено строковое значение '49'. Аналогичный конвейер можно имплементировать на надмножестве языка Pyton Coconut следующим образом:

2 |> ( x -> x + 5) |> ( x -> x * x) |> ( x -> str(x)) |> print

На самом языке Python для этого нужно написать специальную функцию, и, разумеется, это будет функция более высокого порядка с использованием цикла:

# Конвейер обработки данных
def pipe(data, *fseq):
    for fn in fseq: 
        data = fn(data)
    return data

Либо с использованием функции reduce из встроенной библиотеки functools:

# Альтернативная имплементация конвейера
from functools import reduce
def pipe(*args): return reduce(lambda x, f: f(x), args)

Приведенный ниже пример демонстрирует работу конвейера на языке Python:

pipe(2,
     lambda x: x + 5,
     lambda x: x * x,
     lambda x: str(x))

или в более удобном виде:

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

pipe(2,
     add(5),
     square,
     tostring)

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

Перед тем, как рассматривать конвейер обработки данных детально, сначала для сравнения приведу имплементацию конвейера в парадигме объектно-ориентированного программирования (программный код позаимствован отсюда):

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:

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

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

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

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

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

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

def datain():
    return int(input('Введите неотрицательное целое число: '))

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

main()

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

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

Лямбда-функция в последнем узле конвейера (put_data()) получает кортеж, состоящий из введенного пользователем числа и полученного результата.

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

from functools import reduce

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

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 datain():
        def get_int(msg=''):  # Утилитная функция
            return int(input(msg))      
          
        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 dataout():
        def fn(data):
            n, fact = data
            print(f'Факториал числа {n} равняется {fact}') 
        return fn

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

main()

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

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

Функциональным ядром приведенной выше программы являются строки:

pipe(datain(), 
     factorial, 
     dataout())

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

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

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

  • Позволяет менять узлы конвейера на другие с целью тестирования различных и более эффективных имплементаций алгоритмов. Например, вместо нерекурсивной функции factorial, можно поместить рекурсивную функцию factorial_rec.

pipe(datain(), factorial_rec, dataout())
  • Облегчает проведение отладки программы, позволяя на каждом стыке вставлять отладочный код с целью проверки промежуточных результатов и тестирования производительности отдельных узлов.

Например, рассмотрим вторую возможность – отладку. В этом случае можно написать вспомогательную функцию debug:

def debug(data):
    print(data) 
    return data

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

pipe(datain(), debug, factorial, debug, dataout())

Если выполнить программу в таком варианте, то будут получены следующие результаты:

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

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

Факториал числа 4 равняется 24

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

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

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

from functools import reduce

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

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 datain():
        def get_int(msg=''):  # Утилитная функция
            return int(input(msg)) 
          
        def validate(n):  # Валидация входных данных    
            if not isinstance(n, int):
                raise TypeError("Число должно быть целым.")
            if not n >= 0:
                raise ValueError("Число должно быть ноль положительным.")
            if n > 10:
                raise ValueError("Число должно быть не больше 10.")
            return n    
          
        msg = 'Введите неотрицательное целое число не больше 10: '
        return pipe(get_int(msg), validate)

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

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

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 datain():
        seq = [1, 2, 3, 4, 5, 6, 7, 8, 9]    
        params = (2,5)   # params - это параметры start, end
        return seq, params  

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

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

main()

Вывод программы
Сумма значений со 2 по 5 позиции равняется 18

Выводы

Приведенный в настоящей главе материал имеет ознакомительный характер и служит для иллюстрации возможностей функциональной парадигмы программирования на Python с целью дальнейших самостоятельных исследований и чтобы побудить программистов дать функциональному стилю шанс. Помимо конвейера функциональное ядро может быть имплементировано в виде графа: дерева или ориентированного ациклического графа. Сам конвейер может быть расширен изнутри и включать ветвление. Кроме того, конвейеры могут использоваться в разных частях программы, собираться вместе, например, по условию, и пр. Как всегда, все зависит от потребности и фантазии разработчика.) Например, конвейер можно имплементировать в составе функции загрузки данных:

# Множественная диспетчеризация (мультиметод)

import pandas as pd

class Data:
    uk, uk_scrbd, ru = range(3)
    
def load_data(identity):
    '''имплементация мультиметода на Python; загружает
    данные в зависимости от значения идентификатора'''
    return {
        Data.uk: lambda: do('ch01/UK2010.xls', 
                            pd.read_excel 
                           ),
        Data.ru: lambda: do('ch01/RU2011.xls', 
                            pd.read_excel, 
                            lambda o: o[o['Election Year'].notnull()]
                           )
    }[identity]()

load_data(Data.uk)

Исходный код поста находится в моем репо на Github. Материал поста использовался в качестве авторского дополнения в русском переводе книги «Strating Out with Python». В англоязычном интернете можно найти материал, в котором рассматриваются различные варианты имплементации конвейера на Python и шаблоны ветвления потока данных внутри конвейера. Например,

  • Доклад на конференции PyCon, в котором обсуждается архитектурный шаблон «функциональное ядро и императивная оболочка».

  • Еще один доклад на конференции PyCon, в котором на 11-ой минуте обсуждается та же тематика.

  • Очень рекомендую презентацию на Youtube «Конвейеризация на Python - конвейеры в приложениях науки о данных».

  • Несколько общих шаблонов дизайна конвейеров можно найти здесь, здесь и здесь.

Если же помимо функционала map/filter/reduce/zip и библиотеки functools вам нужны более функциональные инструменты, то существуют сторонние пакеты Python, такие как:

  • Пакет pyrsistent, который предлагает немутируемые структуры данных

  • Пакет pydash, который предлагает функциональные инструменты

  • Пакет fnc, который предлагает функциональные инструменты

  • Пакет toolz, который предлагает стандартную библиотеку ФП

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

Только зарегистрированные пользователи могут участвовать в опросе. Войдите, пожалуйста.
Полезность материала
31.58% Высокая 12
18.42% Нейтральная 7
50% Низкая 19
Проголосовали 38 пользователей. Воздержались 12 пользователей.
Теги:
Хабы:
Если эта публикация вас вдохновила и вы хотите поддержать автора — не стесняйтесь нажать на кнопку
-7
Комментарии 78
Комментарии Комментарии 78

Публикации

Истории

Работа

Python разработчик
128 вакансий
Data Scientist
58 вакансий

Ближайшие события

Московский туристический хакатон
Дата 23 марта – 7 апреля
Место
Москва Онлайн