Fugue, Rapids, BlazingSQL integration

This project extends Fugue to support Rapids cuDF and BlazingSQL.

Installation

You need to install Rapids and BlazingSQL by yourself (see official instructions), and assume you installed them by conda, then you need to pip install in the same environment

conda run -n <your_env> pip install fugue-blazing

How To Use

As a standard Fugue extension, you can use in two ways: functional APIs and Fugue SQL. But Fugue SQL is the preferred way for this extension. This is because due to the special design of GPU, code to run on GPU has special requirement. Currently transform is leveraging NativeExecutionEngine which is using CPU. Other than transform, Fugue fully relies on cuDF and BlasingSQL to do the compute.

Practically, if you don't use transform, then SQL may be the better choice to express your data pipelines.

Functional APIs

Here is an example Fugue code snippet that illustrates some of the key features of the framework. A fillna function creates a new column named filled, which is the same as the column value except that the None values are filled.

from fugue import FugueWorkflow
from fugue_blazing import CudaExecutionEngine, setup_shortcuts

# Creating sample data
data = [
    ["A", "2020-01-01", 10],
    ["A", "2020-01-02", None],
    ["A", "2020-01-03", 30],
    ["B", "2020-01-01", 20],
    ["B", "2020-01-02", None],
    ["B", "2020-01-03", 40]
]
schema = "id:str,date:date,value:double"

dag = FugueWorkflow()
dag.df(data, schema).partition_by("id", presort="date").take(1).show()

dag.run(CudaExecutionEngine)

# call setup_shortcuts to make your code more expressive
setup_shortcuts()
dag.run("blazing")

You can also run SQL using functional API:

from fugue import FugueWorkflow
from fugue_blazing import setup_shortcuts

setup_shortcuts()

data = [
    ["A", "2020-01-01", 10],
    ["A", "2020-01-02", None],
    ["A", "2020-01-03", 30],
    ["B", "2020-01-01", 20],
    ["B", "2020-01-02", None],
    ["B", "2020-01-03", 40]
]
schema = "id:str,date:date,value:double"

with FugueWorkflow("blazing") as dag:
    df = dag.df(data, schema)
    dag.select("* from ",df," where value>20").show()

For detailed examples, please read Fugue Tutorials

Fugue SQL

Programmatical Approach

from fugue_sql import fsql
from fugue_blazing import setup_shortcuts
import pandas as pd
import cudf

setup_shortcuts()

pdf = pd.DataFrame([
    ["A", "2020-01-01", 10],
    ["A", "2020-01-02", None],
    ["A", "2020-01-03", 30],
    ["B", "2020-01-01", 20],
    ["B", "2020-01-02", None],
    ["B", "2020-01-03", 40]
], columns = ["id", "date", "value"])

result = fsql("""
TAKE 1 ROW FROM df PREPARTITION BY id PRESORT date
YIELD DATAFRAME AS x
""", df=pdf).run("blazing")

# this is how you get outputs from Fugue SQL
assert isinstance(result["x"].native, cudf.DataFrame)

fsql("""
SELECT * FROM best WHERE id='A'
PRINT
SELECT id, COUNT(*) AS ct FROM orig GROUP BY id
PRINT
""", best=result["x"], orig=pdf).run("blazing")

Jupyter Notebook

Before running Jupyter, you need to firstly install fugue and notebook extension

pip install fugue
jupyter nbextension install --sys-prefix --symlink --py fugue_notebook
jupyter nbextension enable --py fugue_notebook

In cell 1

%load_ext fugue_notebook

from fugue_blazing import setup_shortcuts
setup_shortcuts()

pdf = pd.DataFrame([
    ["A", "2020-01-01", 10],
    ["A", "2020-01-02", None],
    ["A", "2020-01-03", 30],
    ["B", "2020-01-01", 20],
    ["B", "2020-01-02", None],
    ["B", "2020-01-03", 40]
], columns = ["id", "date", "value"])

In cell 2

%%fsql blazing
TAKE 1 ROW FROM df PREPARTITION BY id PRESORT date
YIELD DATAFRAME AS x

In cell 3

%%fsql blazing
SELECT * FROM x WHERE id='A'
PRINT
SELECT id, COUNT(*) AS ct FROM pdf GROUP BY id
PRINT