fugue

教程	API 文档	在 Slack 上与我们交流！

Fugue 是一个统一的分布式计算接口，允许用户在 Spark、Dask 和 Ray 上执行 Python、Pandas 和 SQL 代码，只需进行最小程度的重写。

Fugue 最常用于：

并行化或扩展现有的 Python 和 Pandas 代码，通过最小程度的重写将其迁移到 Spark、Dask 或 Ray。
使用 FugueSQL 定义端到端工作流，在 Pandas、Spark 和 Dask DataFrames 之上。FugueSQL 是一个增强的 SQL 接口，可以调用 Python 代码。

要了解 Fugue 与其他框架（如 dbt、Arrow、Ibis、PySpark Pandas）的比较，请参阅对比

Fugue API

Fugue API 是一组能够在 Pandas、Spark、Dask 和 Ray 上运行的函数集合。使用 Fugue 最简单的方法是 transform() 函数。这让用户可以通过将单个函数迁移到 Spark、Dask 或 Ray 来实现并行执行。在下面的示例中，map_letter_to_food() 函数接受一个映射并将其应用于一列。到目前为止，这只是 Pandas 和 Python（不包括 Fugue）。

import pandas as pd
from typing import Dict

input_df = pd.DataFrame({"id":[0,1,2], "value": (["A", "B", "C"])})
map_dict = {"A": "Apple", "B": "Banana", "C": "Carrot"}

def map_letter_to_food(df: pd.DataFrame, mapping: Dict[str, str]) -> pd.DataFrame:
    df["value"] = df["value"].map(mapping)
    return df

现在，通过调用 Fugue 的 transform() 函数，map_letter_to_food() 函数被迁移到 Spark 执行引擎。输出 schema 和 params 被传递给 transform() 调用。schema 是必需的，因为这是分布式框架的要求。下面的 schema="*" 表示所有输入列都在输出中。

from pyspark.sql import SparkSession
from fugue import transform

spark = SparkSession.builder.getOrCreate()
sdf = spark.createDataFrame(input_df)

out = transform(sdf,
               map_letter_to_food,
               schema="*",
               params=dict(mapping=map_dict),
               )
# out 是一个 Spark DataFrame
out.show()

+---+------+
| id| value|
+---+------+
|  0| Apple|
|  1|Banana|
|  2|Carrot|
+---+------+

Fugue transform() 的 PySpark 等效代码

from typing import Iterator, Union
from pyspark.sql.types import StructType
from pyspark.sql import DataFrame, SparkSession

spark_session = SparkSession.builder.getOrCreate()

def mapping_wrapper(dfs: Iterator[pd.DataFrame], mapping):
  for df in dfs:
      yield map_letter_to_food(df, mapping)

def run_map_letter_to_food(input_df: Union[DataFrame, pd.DataFrame], mapping):
  # 转换
  if isinstance(input_df, pd.DataFrame):
      sdf = spark_session.createDataFrame(input_df.copy())
  else:
      sdf = input_df.copy()

  schema = StructType(list(sdf.schema.fields))
  return sdf.mapInPandas(lambda dfs: mapping_wrapper(dfs, mapping),
                          schema=schema)

result = run_map_letter_to_food(input_df, map_dict)
result.show()

这种语法比 PySpark 等效代码更简单、更清晰、更易于维护。同时，为了将原始的基于 Pandas 的函数迁移到 Spark，没有对其进行任何修改。它仍然可以用于 Pandas DataFrames。Fugue transform() 还支持 Dask 和 Ray 作为执行引擎，以及默认的基于 Pandas 的引擎。

Fugue API 有更广泛的函数集合，这些函数也与 Spark、Dask 和 Ray 兼容。例如，我们可以使用 load() 和 save() 来创建与 Spark、Dask 和 Ray 兼容的端到端工作流。有关完整的函数列表，请参阅顶级 API

import fugue.api as fa

def run(engine=None):
    with fa.engine_context(engine):
        df = fa.load("/path/to/file.parquet")
        out = fa.transform(df, map_letter_to_food, schema="*")
        fa.save(out, "/path/to/output_file.parquet")

run()                 # 在 Pandas 上运行
run(engine="spark")   # 在 Spark 上运行
run(engine="dask")    # 在 Dask 上运行

上下文中的所有函数都将在指定的后端上运行。这使得在本地执行和分布式执行之间切换变得容易。

FugueSQL

FugueSQL 是一种基于 SQL 的语言，能够在 Pandas、Spark 和 Dask 之上表达端到端的数据工作流。上面的 map_letter_to_food() 函数在下面的 SQL 表达式中使用。这展示了如何将 Python 定义的函数与标准 SQL SELECT 语句一起使用。

from fugue.api import fugue_sql
import json

query = """
    SELECT id, value
      FROM input_df
    TRANSFORM USING map_letter_to_food(mapping={{mapping}}) SCHEMA *
    """
map_dict_str = json.dumps(map_dict)

# 返回 Pandas DataFrame
fugue_sql(query,mapping=map_dict_str)

# 返回 Spark DataFrame
fugue_sql(query, mapping=map_dict_str, engine="spark")

安装

Fugue 可以通过 pip 或 conda 安装。例如：

pip install fugue

为了使用 Fugue SQL，强烈建议安装 sql 额外组件：

pip install fugue[sql]

它还有以下安装额外组件：

sql：支持 Fugue SQL。没有这个额外组件，非 SQL 部分仍然可以工作。在 Fugue 0.9.0 之前，这个额外组件包含在 Fugue 的核心依赖中，所以你不需要显式安装。但对于 0.9.0+，如果你想使用 Fugue SQL，这就变成必需的了。
spark：支持 Spark 作为 ExecutionEngine。
dask：支持 Dask 作为 ExecutionEngine。
ray：支持 Ray 作为 ExecutionEngine。
duckdb：支持 DuckDB 作为 ExecutionEngine，阅读详情。
polars：支持 Polars DataFrames 和使用 Polars 的扩展。
ibis：为 Fugue 工作流启用 Ibis，阅读详情。
cpp_sql_parser：为 Fugue SQL 启用 CPP antlr 解析器。它可以比纯 Python 解析器快 50 多倍。对于主要的 Python 版本和平台，已经有预构建的二进制文件，但对于其余的，它需要一个 C++ 编译器来即时构建。

例如，一个常见的用例是：

pip install "fugue[duckdb,spark]"

注意，如果你已经独立安装了 Spark 或 DuckDB，Fugue 能够自动使用它们，而不需要安装额外组件。

入门

开始使用 Fugue 的最佳方式是完成 10 分钟教程：

对于顶级 API，请参阅：

Fugue 顶级 API

教程也可以通过 binder 或 Docker 在交互式笔记本环境中运行：

使用 binder

注意在 binder 上运行速度较慢，因为 binder 上的机器对于分布式框架（如 Spark）来说不够强大。并行执行可能变成顺序执行，所以一些性能比较的例子可能无法给你正确的数据。

使用 Docker

另外，你可以通过在自己的机器上运行这个 Docker 镜像来获得不错的性能：

docker run -p 8888:8888 fugueproject/tutorials:latest

Jupyter Notebook 扩展

有一个配套的 notebook 扩展用于 FugueSQL，让用户可以使用 %%fsql 单元格魔法。该扩展还为 FugueSQL 单元格提供语法高亮。它适用于经典 notebook 和 Jupyter Lab。更多详情可以在安装说明中找到。

FugueSQL gif

生态系统

作为一个抽象层，Fugue可以与许多其他开源项目无缝集成。

Python后端:

Pandas
Polars（仅限DataFrame）
Spark
Dask
Ray
Ibis

FugueSQL后端:

Pandas - FugueSQL可以在Pandas上运行
Duckdb - 进程内SQL OLAP数据库管理
dask-sql - Dask的SQL接口
SparkSQL
BigQuery
Trino

Fugue可作为后端使用或与以下项目集成:

WhyLogs - 数据分析
PyCaret - 低代码机器学习
Nixtla - 时间序列建模
Prefect - 工作流编排
Pandera - 数据验证
Datacompy（由Capital One提供） - 比较DataFrame

已注册的第三方扩展（主要用于Fugue SQL）包括:

Pandas plot - 使用matplotlib或plotly可视化数据
Seaborn - 使用seaborn可视化数据
WhyLogs - 可视化数据分析
Vizzu - 使用ipyvizzu可视化数据

社区和贡献

欢迎在Slack上与我们联系。我们还提供了贡献指南。

案例研究

LyftLearn如何通过Kubernetes Spark和Fugue实现分布式计算的民主化
Clobotics - 通过Fugue使用Spark进行大规模图像处理
使用Delta Lake、Fugue和Spark构建数据湖REST API的架构（由bitsofinfo撰写的文章）

提及的用途

在Interos, Inc.将数据科学投入生产（Anthony Holten的LinkedIn帖子）
在贝恩公司使用Fugue和Nixtla进行多时间序列预测（Fahad Akbar的LinkedIn帖子）