10分钟快速上手Fugue项目中的FugueSQL

什么是FugueSQL

FugueSQL是Fugue项目的SQL接口，旨在为数据工程师和分析师提供一个简单统一的SQL接口，能够在Pandas、Spark和Dask等不同计算引擎上执行。对于熟悉SQL但需要处理大规模数据的用户来说，FugueSQL提供了一种无缝过渡到分布式计算的方式。

安装与设置

基础安装

要使用FugueSQL，首先需要安装核心包：

pip install fugue[sql]

Jupyter扩展

为了在Jupyter Notebook或JupyterLab中获得更好的开发体验（如语法高亮），还需要安装扩展：

pip install fugue-jupyter
fugue-jupyter install startup

安装完成后，在Jupyter中可以直接使用%%fsql魔法命令来编写FugueSQL代码。

基本语法特性

标准SQL兼容性

FugueSQL完全兼容标准SQL语法，可以直接执行常规的SELECT、JOIN、WHERE等操作。例如：

import pandas as pd
df = pd.DataFrame({"col1": ["A","A","A","B","B","B"], "col2": [1,2,3,4,5,6]})
df2 = pd.DataFrame({"col1": ["A", "B"], "col3": [1, 2]})

%%fsql
SELECT df.col1, df.col2, df2.col3
  FROM df
  LEFT JOIN df2
    ON df.col1 = df2.col1
 WHERE df.col1 = "A"
 PRINT

变量赋值与简化语法

FugueSQL引入了变量赋值功能，可以显著减少SQL中的重复代码：

%%fsql
df = LOAD "/tmp/df.parquet"

max_vals = SELECT col1, MAX(col2) AS max_val
            FROM df
        GROUP BY col1

SELECT df.col1, 
       df.col2 / max_vals.max_val AS normalized
  FROM df
  JOIN max_vals
    ON df.col1 = max_vals.col1
 PRINT

匿名表达式

FugueSQL支持匿名表达式，可以进一步简化代码：

%%fsql
LOAD "/tmp/df.parquet"

SELECT col1, MAX(col2) AS max_val
 GROUP BY col1
 PRINT

高级功能

与Python代码集成

FugueSQL可以直接调用Python函数，实现SQL难以表达的复杂逻辑：

# schema: *+col2:float
def std_dev(df: pd.DataFrame) -> pd.DataFrame:
    return df.assign(col2=df['col2']/df['col2'].max())

%%fsql
LOAD "/tmp/df.parquet"
PREPARTITION BY col1
TRANSFORM std_dev
PRINT

文件操作

FugueSQL支持直接读写文件：

%%fsql
df = LOAD "/tmp/df.parquet"
df2 = LOAD "/tmp/df2.parquet"

new = SELECT df.col1, df.col2, df2.col3
        FROM df
        LEFT JOIN df2
          ON df.col1 = df2.col1
       WHERE df.col1 = "A"

SAVE OVERWRITE "/tmp/res.parquet"

适用场景

FugueSQL特别适合以下场景：

1. SQL用户需要处理超出单机内存的数据量
2. 需要在不同计算引擎(Pandas/Spark/Dask)间切换
3. SQL逻辑需要与Python函数结合
4. 简化复杂SQL查询的编写

总结

FugueSQL通过扩展标准SQL语法，提供了更简洁的数据处理方式，同时保持了与Python生态的无缝集成。对于习惯使用SQL的数据从业者，FugueSQL可以大大降低学习分布式计算的曲线，提高开发效率。

通过本教程，您已经掌握了FugueSQL的基本用法，可以开始在实际项目中尝试使用。随着熟练度的提高，可以进一步探索FugueSQL更高级的特性，如分布式执行、自定义函数等。