В приведенном ниже примере также показано, как разделить кадр данных перед применением к нему преобразователя . Это применит преобразователь к каждому разделу.

 при вводе import Iterable, Dict, Any, List
импортировать панд как pd
из импорта фуги FugueWorkflow
def add(df:pd.DataFrame, n=1) -> pd.DataFrame:
    дф["б"]+=n
    вернуть дф
    
def get_top(df:Iterable[Dict[str,Any]]) -> Iterable[Dict[str,Any]]:
    выход следующий (df)
    возвращаться
с FugueWorkflow() как dag:
    df = dag.df([[0,1],[0,2],[1,3],[1,1]],"a:int,b:int")
    # без подсказки схемы вы должны указать схему в коде Fugue
    df = df.  transform (добавить, схема = "*"). Transform (добавить, схема = "*", params = dict (n = 2))
    # получить наименьшее b каждого раздела
    df.partition(by=["a"], presort="b").transform(get_top, schema="*").show()
    # получить наибольшее значение b каждого раздела
    df.partition(by=["a"], presort="b DESC").transform(get_top, schema="*").show()
 

 Пандасдатафраме
а: интервал | б: интервал
-----+-----
0 |4
1 |4
Всего: 2
ПандасДатаФраме
а: интервал | б: интервал
-----+-----
0 |5
1 |6
Всего: 2
 

Подсказка схемы

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

 # схема: *
def add(df:pd.DataFrame, n=1) -> pd.DataFrame:
    дф["б"]+=n
    вернуть дф
    
# схема: *
def get_top(df:Iterable[Dict[str,Any]]) ->
  Iterable[Dict[str,Any]]:
    выход следующий (df)
    возвращаться
с FugueWorkflow() как dag:
    df = dag. df([[0,1],[0,2],[1,3],[1,1]],"a:int,b:int")
    # синтаксис для установки параметров
    df = df.transform(добавить).transform(добавить, params=dict(n=2))
    df.partition(by=["a"], presort="b").transform(get_top).show()
 

 Пандасдатафраме
а: интервал | б: интервал
-----+-----
0 |4
1 |4
Всего: 2
 

Синтаксис подсказки схемы

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

 # схема: *,c:int
def with_c(df:pd.DataFrame) -> pd.DataFrame:
    дф["с"]=1
    вернуть дф
# схема: *-b
def drop_b(df:pd.DataFrame) -> pd.DataFrame:
    вернуть df.drop («b», ось = 1)
# схема: *~b,c
def drop_b_c_if_exists(df:pd.DataFrame) ->  pd.DataFrame:
    вернуть df.drop(["b","c"], ось=1, ошибки='игнорировать')
с FugueWorkflow() как dag:
    df = dag.df([[0,1],[0,2]],"a:int,b:int")
    df = df.transform(with_c)
    df.show ()
    df = df. transform (drop_b)
    df.show ()
    df = df.transform (drop_b_c_if_exists)
    df.show ()
 

 Пандасдатафраме
а: интервал | б: интервал | с: интервал
-----+-----+-----
0 |1 |1
0 |2 |1
Всего: 2
ПандасДатаФраме
а: интервал | с: интервал
-----+-----
0 |1
0 |1
Всего: 2
ПандасДатаФраме
а:инт
-----
0
0
Всего: 2
 

Подход декоратора

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

 от импортного трансформатора fugue
# df — это заархивированные кадры данных, **kwargs — это параметры, переданные из Fugue
# приведенный ниже синтаксис эквивалентен @transformer("*,c:int")
@transformer(лямбда df, **kwargs: df.schema+"c:int")
def with_c(df:pd.DataFrame) -> pd.DataFrame:
    дф["с"]=1
    вернуть дф
с FugueWorkflow() как dag:
    df = dag.df([[0,1],[0,2],[1,3],[1,1]],"a:int,b:int")
    df = df. transform(with_c)
    df.show ()
 

 Пандасдатафраме
а: интервал | б: интервал | с: интервал
-----+-----+-----
0 |1 |1
0 |2 |1
1 |3 |1
1 |1 |1
Всего: 4
 

Подход к интерфейсу (расширенный)

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

• Для выходной схемы требуется информация о разделе, такая как ключи раздела, схема и текущие значения ключей.

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

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

С этого момента мы будем использовать функцию create_helper , которая создает для нас случайный кадр данных pandas.

 импортировать numpy как np
def create_helper(ct=20) -> pd.DataFrame:
    np.random.seed (0)
    вернуть pd.DataFrame(np.random.randint(0,10,size=(ct, 3)), столбцы=список('abc'))
 

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

 из фуги импортировать Transformer, PandasDataFrame, DataFrame, LocalDataFrame
из времени импортировать сон
деф дорогая_инит (сек = 5):
    сон (сек)
класс медиана (преобразователь):
    # вызывается на стороне драйвера
    защита get_output_schema (я, дф):
        вернуть df. schema + (self.params.get_or_throw ("col", str), float)
    
    # при инициализации физического раздела
    def on_init(self, df: DataFrame) -> Нет:
        self.col = self.params.get_or_throw("col", str)
        дорогой_init(self.params.get("сек",0))
        
    преобразование защиты (я, дф):
        pdf = df.as_pandas()
        pdf[self.col]=pdf["b"].медиана()
        вернуть PandasDataFrame (pdf)
        
с FugueWorkflow() как dag:
    df = dag.create(create_helper)
    df.partition(by=["a"]).transform(Median, params={"col":"med", "sec": 1}).show(rows=5)
 

 Пандасдатафраме
a:длинный|b:длинный|c:длинный|средний:двойной
------+------+------+----------
0 |4 |7 |4,0
1 |3 |3 |4.0
1 |9 |9 |4.0
1 |4 |9 |4.0
2 |3 |8 |1,5
 

В качестве примечания, этот пример показывает, что параметры могут быть получены с помощью self.params.get или self.params.get_or_throw . self.params — это словарь, поэтому метод get аналогичен доступу к словарю. self.params.get_or_throw выдает ошибку, если параметр не соответствует заданному типу.

Чтобы показать преимущества on_init , мы также создали другую версию преобразователя Median , используя подсказку схемы. Это также вызывает в этой функции для каждого логического раздела. Также в функции запуска мы установили num=2 , чтобы показать эффект при использовании 2 воркеров. Таким образом, для преобразователя Median , который использовал интерфейс, дорогая_инит будет вызываться не более двух раз, но для версии, которая использовала подсказку схемы, она будет вызываться больше раз.

Цифры могут быть неверными, если вы запустите это на связующем, но обратите внимание на разницу в величине.

 из импорта fugue_spark SparkExecutionEngine
из timeit импортировать timeit
# схема: *, m:двойной
def median(df:pd.DataFrame, sec=0) -> pd.DataFrame:
    дорогая_инициализация (сек)
    df["m"]=df["b"]. медиана()
    вернуть дф
def run(двигатель, без интерфейса, сек):
    с FugueWorkflow (движок) как dag:
        df = dag.create(create_helper)
        если без интерфейса:
            df.partition(by=["a"], num=2).transform(median, params={"sec": sec}).show(rows=5)
        еще:
            df.partition(by=["a"], num=2).transform(Median, params={"col":"m", "sec": sec}).show(rows=5)
    
двигатель = SparkExecutionEngine()
print(f"Время выполнения без интерфейса: {timeit(lambda: run(engine, True, 1), number=1)}")
print(f"Время выполнения интерфейса: {timeit(lambda: run(engine, False, 1), number=1)}")
 

 Спаркдатафрейм
a:длинный|b:длинный|c:длинный|m:двойной
------+------+------+--------
2 |3 |8 |1,5
2 |0 |0 |1,5
4 |7 |6 |5.0
4 |3 |0 |5,0
4 |5 |5 |5,0
Время выполнения без интерфейса: 10.233126380999238
SparkDataFrame
a:длинный|b:длинный|c:длинный|m:двойной
------+------+------+--------
2 |3 |8 |1,5
2 |0 |0 |1,5
4 |7 |6 |5.0
4 |3 |0 |5,0
4 |5 |5 |5.0
Время выполнения интерфейса: 1,8927552820005076
 

Использование метода on_init значительно ускорило операцию, потому что дорогие_инициалы не повторялись без необходимости.

Преобразование фуги

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

Функция преобразования полезна для распараллеливания одной функции над Spark или Dask.

 # схема не указана, поэтому она будет передана позже
def median1(df:pd.DataFrame, sec=0) -> pd.DataFrame:
    df["m"]=df["b"].медиана()
    вернуть дф
# схема: *, m:двойной
def median2(df:pd.DataFrame, sec=0) -> pd.DataFrame:
    df["m"]=df["b"].медиана()
    вернуть дф
 

Затем можно использовать эти два трансформатора. median1 было записано с помощью Native Approach , поэтому схема потребуется для вызова transform ниже. median2 , с другой стороны, использует подсказку схемы для предоставления схемы, поэтому ее не нужно предоставлять во время выполнения.

В обоих приведенных ниже случаях мы передаем SparkExecutionEngine , который преобразует исходный df в Spark DataFrame и выполняет медианных функций распределенным способом. Это также возвращает Spark DataFrame, поскольку вычисление выполнялось на SparkExecutionEngine . Его можно преобразовать обратно в pandas , используя метод toPandas() Spark DataFrames, но этот метод эффективен только для небольших данных.

 из преобразования импорта фуги
# образец кадра данных pandas
ДФ = создать_помощник ()
df1 = преобразование (df,
               медиана,
               схема = "*, м: двойной",
               двигатель = SparkExecutionEngine,
               раздел = дикт (by = "а")
               )
# схема известна для median2
df2 = преобразование (df,
               медиана2,
               двигатель = SparkExecutionEngine,
               раздел = дикт (by = "а")
               )
df1. show(2)
# преобразовать обратно в панд
df2.toPandas().голова(2)
 

 +---+---+---+---+
| а| б| с | м|
+---+---+---+---+
| 0| 4| 7|4.0|
| 6| 7| 7|7,5|
+---+---+---+---+
показаны только верхние 2 строки
 

BP | Бумаги с кодом

Или обсудите изменение в Slack .

**BP-Transformer (BPT)** — это тип [Transformer](https://paperswithcode. com/method/transformer), который мотивирован необходимостью найти лучший баланс между возможностями и вычислительной сложностью для самоконтроля. внимание. Архитектура разделяет входную последовательность на разные многомасштабные диапазоны с помощью двоичного разделения (BP). Он включает в себя индуктивное смещение внимания к контекстной информации от мелкозернистой к крупнозернистой по мере увеличения относительного расстояния. Чем дальше контекстная информация, тем грубее ее представление. BPT можно рассматривать как графовую нейронную сеть, узлами которой являются разномасштабные отрезки. Маркерный узел может обслуживать диапазон меньшего масштаба для более близкого контекста и диапазон большего масштаба для контекста большего расстояния. Представления узлов обновляются с помощью [Graph Self-Attention] (https://paperswithcode.com/method/graph-self-attention).

Прикрепленные коллекции:

• ТРАНСФОРМАТОРЫ

---------