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

Fugue是一个统一的分布式计算接口，允许用户在Spark、Dask和Ray上运行Python、Pandas和SQL代码，且只需进行最少的修改。

Fugue最常用于：

• 并行化或扩展现有的Python和Pandas代码，通过将其移植到Spark、Dask或Ray上，而无需大量重写。
• 使用FugueSQL来定义基于Pandas、Spark和Dask DataFrame的端到端工作流。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是因为这是分布式框架的要求。下面的"*"表示输入的所有列都会出现在输出中。

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|
+---+------+

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 DataFrame上使用。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]

此外，Fugue 还提供了以下安装附加组件：

• 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 及其扩展功能。
• ibis: 用于在 Fugue 工作流中启用 Ibis，请参阅 详细信息。
• cpp_sql_parser: 用于启用 Fugue SQL 的 CPP antlr 解析器。它比纯 Python 解析器快 50 倍以上。对于主流的 Python 版本和平台，已经提供了预编译的二进制文件，但对于其他情况，则需要 C++ 编译器进行即时编译。

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

pip install "fugue[duckdb,spark]"

请注意，如果您已经独立安装了 Spark 或 DuckDB，Fugue 会自动使用它们，而无需安装这些附加组件。

入门指南

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

• 10 分钟学会 Fugue API
• 10 分钟学会 FugueSQL

有关顶层 API 的更多信息，请参阅：

• Fugue 顶层 API

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

使用 binder

请注意，在 binder 上运行速度较慢，因为 binder 上的机器性能不足以支持像 Spark 这样的分布式框架。并行执行可能会变为串行执行，因此一些性能对比示例可能无法给出准确的结果。

使用 Docker

或者，您可以在自己的机器上运行以下 Docker 镜像，以获得较好的性能：

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

Jupyter Notebook 扩展

FugueSQL 配备了一个配套的 notebook 扩展，允许用户使用 %%fsql 单元魔法命令。该扩展还为 FugueSQL 单元提供语法高亮显示功能，适用于经典 notebook 和 Jupyter Lab。更多详情请参阅 安装说明。

生态系统

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

Python 后端：

• Pandas
• Polars（仅限 DataFrames）
• 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 开发) - 比较 DataFrames

已注册的第三方扩展（主要用于 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 帖子）
• Bain & Company 使用 Fugue 和 Nixtla 进行多时间序列预测（Fahad Akbar 的 LinkedIn 帖子）

更多资源

查看我们最新的会议演讲和相关内容。如需更完整的列表，请访问教程中的 内容 页面。

博客

• 为什么类 Pandas 的接口对分布式计算并不理想
• 介绍 FugueSQL — 适用于 Pandas、Spark 和 Dask DataFrame 的 SQL（Khuyen Tran 在 Towards Data Science 上的文章）

会议

• Lyft 公司的分布式机器学习
• 比较扩展 Python 和 Pandas 代码的不同方法
• 使用 Spark 和 Dask 进行大规模数据验证（PyCon US）
• FugueSQL — 面向 Pandas、Spark 和 Dask DataFrame 的增强型 SQL 接口（PyData Global）
• 分布式混合参数调优

Fugue 快速上手指南

Fugue 是一个统一的分布式计算接口，允许用户使用极少的代码修改，将 Python、Pandas 和 SQL 代码在 Spark、Dask 和 Ray 等引擎上运行。它特别适合将现有的单机 Pandas 代码快速扩展为分布式任务。

环境准备

• 操作系统：Linux, macOS, Windows
• Python 版本：支持 Python 3.8+ (具体版本请参考 PyPI 徽章)
• 前置依赖：
  • 基础使用仅需 Python 环境。
  • 若需使用特定分布式引擎（如 Spark, Dask, Ray），需确保本地已安装相应环境或通过网络可访问集群。
  • 若需使用 FugueSQL，建议安装额外的 SQL 解析支持。

安装步骤

1. 基础安装

通过 pip 安装核心库：

pip install fugue

国内加速建议：如果下载速度慢，可使用清华或阿里镜像源：

pip install fugue -i https://pypi.tuna.tsinghua.edu.cn/simple

2. 按需安装扩展功能

根据你需要的执行引擎或功能安装额外组件：

• 启用 FugueSQL (强烈推荐)：

  pip install "fugue[sql]"
  

• 启用 Spark 支持：

  pip install "fugue[spark]"
  

• 启用 Dask 支持：

  pip install "fugue[dask]"
  

• 启用 DuckDB 支持 (高性能本地 SQL)：

  pip install "fugue[duckdb]"
  

• 组合安装示例 (同时支持 SQL 和 Spark)：

  pip install "fugue[sql,spark]"
  

注意：如果你已经独立安装了 Spark 或 DuckDB，Fugue 通常能自动识别并使用它们，无需安装对应的 extras，但安装 extras 可确保依赖兼容性。

基本使用

Fugue 的核心优势在于无需修改原有的 Pandas 函数即可将其转换为分布式任务。以下是最简单的 transform 用法示例。

场景：将 Pandas 函数并行化

假设你有一个标准的 Pandas 处理函数：

import pandas as pd
from typing import Dict

# 1. 定义原始 Pandas 函数
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

2. 使用 Fugue 切换到分布式引擎

只需调用 fugue.transform，指定目标引擎（如 spark），即可在分布式环境下运行上述函数，而无需重写任何逻辑。

from pyspark.sql import SparkSession
from fugue import transform

# 初始化 Spark Session
spark = SparkSession.builder.getOrCreate()
sdf = spark.createDataFrame(input_df)

# 使用 Fugue transform 执行
# schema="*" 表示输出保留所有输入列结构
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|
+---+------+

3. 灵活切换执行后端

使用 engine_context 可以方便地在本地 (Pandas) 和分布式 (Spark/Dask) 之间切换，便于开发和调试：

import fugue.api as fa

def run_workflow(engine=None):
    with fa.engine_context(engine):
        # 下面的代码会根据 context 自动在指定引擎上运行
        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_workflow()                 # 默认在 Pandas 上运行
run_workflow(engine="spark")   # 在 Spark 上运行
run_workflow(engine="dask")    # 在 Dask 上运行

通过以上步骤，你可以轻松地将现有的数据处理逻辑扩展到大规模分布式集群。更多高级用法（如 FugueSQL）请参考官方教程。