# Prefect Fugue Integration
[](https://GitHub.com/fugue-project/prefect-fugue)
[](https://pypi.python.org/pypi/prefect-fugue/)
[](https://pypi.python.org/pypi/prefect-fugue/)
[](https://pypi.python.org/pypi/prefect-fugue/)
[](https://codecov.io/gh/fugue-project/prefect-fugue)
[](https://prefect-fugue.readthedocs.org)
[](http://slack.fugue.ai)
This project provides the Fugue tasks, context and blocks for Prefect.
## Getting Started
Fugue is a unified interface for distributed computing that lets users execute Python, pandas, and SQL code on Spark, Dask and Ray without rewrites.
The most common use cases are:
* Accelerating or scaling existing Python and pandas code by bringing it to Spark or Dask with minimal rewrites.
Using FugueSQL to define end-to-end workflows on top of pandas, Spark, and Dask DataFrames. FugueSQL is an enhanced SQL interface that can invoke Python code with added keywords.
* Maintaining one codebase for pandas, Spark, Dask and Ray projects. Logic and execution are decoupled through Fugue, enabling users to be focused on their business logic rather than writing framework-specific code.
* Improving iteration speed of big data projects. Fugue seamlessly scales execution to big data after local development and testing. By removing PySpark code, unit tests can be written in Python or pandas and ran locally without spinning up a cluster.
The best way to get started with Fugue is to work through the 10 minute tutorials:
* [Fugue in 10 minutes](https://fugue-tutorials.readthedocs.io/tutorials/quick_look/ten_minutes.html)
* [FugueSQL in 10 minutes](https://fugue-tutorials.readthedocs.io/tutorials/quick_look/ten_minutes_sql.html)
### Python setup
Requires an installation of Python 3.8+.
We recommend using a Python virtual environment manager such as pipenv, conda or virtualenv.
### Installation
Install `prefect-fugue` with `pip`:
```bash
pip install prefect-fugue
```
It's also recommended to register Fugue blocks into your current Prefect workspace:
```bash
prefect block register -m prefect_fugue
```
### Creating a Block
This will allow a creation of the Fugue Engine block
There are 4 items that need to be filled to create a block.
* Block Name - name that will be used to use the block.
* Engine Name - one of the Fugue supported backends (spark, dask, ray, duckdb)
* Engine Config - configurations related to the cluster
* Secret Config - credentials to connect to a cluster
For example, a Databricks Block could look like:
* Block Name - databricks
* Engine Name - spark
* Engine Config - None
* Secret Config - seen below
```json
{
"host": "https://dbc-38aaa459-faaf.cloud.databricks.com",
"token": "dapiecaaae64a727498daaaaafe1bace968a",
"cluster_id": "0612-191111-6fopaaaa"
}
```
As long as you installed `prefect_fugue`, Fugue is able to recognize and convert a block expression to a `FugueExecutionEngine`. For example if you have a block with path `fugue/databricks`, then the expression `prefect:fugue/databricks` becomes a valid execution engine expression. When fugue parse this expression, it will load the `Block` from `fugue/databricks`, then base on the fields of the block, it will construct a `DatabricksExecutionEngine` for your Fugue operations.
### Using a Spark Cluster Inside a Flow
Let’s start by running code on top of Databricks. `databricks-connect` is already installed in this environment. This section may have a lot of logs because of the monitoring provided by Prefect. This section also assumes that the user has Prefect configured to the right workspace.
Below we have one task that takes in a `SparkSession` and uses it to run some Spark code. We can then use this in the Prefect Flow with the `fugue.api.engine_context`. This will create an ephemeral cluster to run the code underneath, and then turn off when finished.
```python
from prefect import task, flow
import fugue.api as fa
@task
def my_spark_task(spark, n=1):
df = spark.createDataFrame([[f"hello spark {n}"]], "a string")
df.show()
@flow
def spark_flow(engine):
with fa.engine_context(engine) as spark_engine:
my_spark_task(spark_engine.spark_session, 1)
spark_flow("prefect:fugue/databricks") # pay attention to the engine expression
```
Similarly, if you don’t use Databricks but have your own way to get a `SparkSession`, you can directly pass the `SparkSession` into the Flow.
```python
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
spark_flow(spark)
```
## More Flexibility
`fugue.api.engine_context` creates a context under which all fugue operations will use the context engine by default. This works either the fugue code is directly under the context or inside the tasks or flows that are invoked under this context.
```python
from prefect import task, flow
import fugue.api as fa
def my_transformer(df:pd.DataFrame) -> pd.DataFrame:
return df
@task
def sub_task(path:str):
df = fa.load(path)
df = fa.transform(df, my_transformer, schema="*")
df.save(path+".output.parquet")
@flow
def sub_flow(path:str):
df = fa.load(path)
df = fa.transform(df, my_transformer, schema="*")
df.save(path+".output.parquet")
@flow
def main_flow(path, engine=None):
with fa.engine_context(engine) as spark_engine:
sub_task(path)
sub_flow(path)
main_flow("<local path>") # test locally, all tasks and flows run without Spark
main_flow("<dbfs path>", "prefect:fugue/databricks") # all tasks and flows will run on Databrickes
```
## Testing Locally Before Running Map Jobs on Spark, Dask, and Ray
We showed how to run Spark code on top of a Spark cluster, but the strength of Fugue is decoupling from distributed framework code such as Spark and Dask. Decoupling from these frameworks allows us to test code locally before scaling out to a cluster. In the example below, we simulate having a pandas DataFrame where each row is a job.
When testing the Flow, we can pass `None` as the engine so everything runs on Pandas. When ready to scale out, we can pass in our `Block` or `SparkSession`. Fugue’s `transform()` task will use the engine provided by the `fugue_engine` context.
```python
from time import sleep
import pandas as pd
import fugue.api as fa
from prefect import task, flow
@task
def create_jobs(n) -> pd.DataFrame:
return pd.DataFrame(dict(jobs=range(n)))
# schema: *,batch_size:str
def run_one_job(df:pd.DataFrame) -> pd.DataFrame:
sleep(len(df)*5)
return df.assign(batch_size=len(df))
@flow
def run_all_jobs(n, engine=None):
jobs = create_jobs(n)
with fa.engine_context(engine):
return transform(jobs, run_one_job, partition="per_row", as_local=True)
```
We can test the Flow above on a local machine without Spark. We run on one job first.
```python
run_all_jobs(1) # run locally on Pandas
```
Becasue it succeeded, we can now attach our Fugue Databricks `Block` to run on Databricks. Now we run on 8 jobs, and we’ll see that parallelization from the Spark cluster will make this Flow execute faster.
```python
run_all_jobs(8, "prefect:fugue/databricks") # run on databricks
```
There is still some overhead with sending the work, but the time is decreased compared to the expected execution time if ran sequentially (40 seconds).
We can also use local Dask by passing the string `"dask"`. We can also pass a `Dask Client()` or use the Fugue Engine `Block` with [Coiled](https://coiled.io/). More information can be found in the [Coiled cloudprovider docs](https://fugue-tutorials.readthedocs.io/tutorials/integrations/cloudproviders/coiled.html).
```python
run_all_jobs(4, dask_client)
```
## Running SQL on any Spark, Dask, and Duckdb
Prototyping locally, and then running the full job on the cluster is also possible with [FugueSQL](https://fugue-tutorials.readthedocs.io/tutorials/quick_look/ten_minutes_sql.html). DuckDB is a good engine to run SQL queries on flat files or Pandas DataFrames. When ready, we can bring it to SparkSQL on the cluster. Similar to the transform() task shown above, there is also an fsql() task.
Here we can load in data and perform a query with FugueSQL. FugueSQL has additional keywords such as LOAD and SAVE so we can run everything from loading, processing, and saving all on DuckDB or SparkSQL. More information on FugueSQL can be found in the FugueSQL tutorials.
```python
import fugue.api as fa
@flow
def run_sql(top, engine):
with fa.engine_context(engine):
fa.fugue_sql_flow("""
df = LOAD "https://d37ci6vzurychx.cloudfront.net/trip-data/green_tripdata_2022-01.parquet"
SELECT PULocationID, COUNT(*) AS ct FROM df
GROUP BY 1 ORDER BY 2 DESC LIMIT {{top}}
PRINT
""", top=top).run()
```
To debug locally without SparkSQL, we can use DuckDB as the engine.
```python
run_sql(2, "duckdb"); # debug/develop without spark
```
Again to run on the cluster, we can use the Databricks Block.
```python
run_sql(10, "prefect:fugue/databricks")
```
## Resources
If you encounter any bugs while using `prefect-fugue`, feel free to open an issue in the [prefect-fugue](https://github.com/fugue-project/prefect-fugue) repository.
If you have any questions or issues while using `prefect-fugue`, you can find help in the [Fugue Slack community](http://slack.fugue.ai).
## Development
If you'd like to install a version of `prefect-fugue` for development, clone the repository and perform an editable install with `pip`:
```bash
git clone https://github.com/fugue-project/prefect-fugue.git
cd prefect-fugue/
pip install -e ".[dev]"
# Install linting pre-commit hooks
pre-commit install
```
Raw data
{
"_id": null,
"home_page": "https://github.com/fugue-project/prefect-fugue",
"name": "prefect-fugue",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "",
"keywords": "prefect",
"author": "The Fugue Development Team",
"author_email": "hello@fugue.ai",
"download_url": "https://files.pythonhosted.org/packages/48/8e/784d45cc3c86b947c0b6a199799036c461025d5015de566c99762b77375c/prefect-fugue-0.0.3.tar.gz",
"platform": null,
"description": "# Prefect Fugue Integration\n\n[](https://GitHub.com/fugue-project/prefect-fugue)\n[](https://pypi.python.org/pypi/prefect-fugue/)\n[](https://pypi.python.org/pypi/prefect-fugue/)\n[](https://pypi.python.org/pypi/prefect-fugue/)\n[](https://codecov.io/gh/fugue-project/prefect-fugue)\n[](https://prefect-fugue.readthedocs.org)\n\n[](http://slack.fugue.ai)\n\n\nThis project provides the Fugue tasks, context and blocks for Prefect.\n\n## Getting Started\n\nFugue is a unified interface for distributed computing that lets users execute Python, pandas, and SQL code on Spark, Dask and Ray without rewrites.\n\nThe most common use cases are:\n\n* Accelerating or scaling existing Python and pandas code by bringing it to Spark or Dask with minimal rewrites.\nUsing FugueSQL to define end-to-end workflows on top of pandas, Spark, and Dask DataFrames. FugueSQL is an enhanced SQL interface that can invoke Python code with added keywords.\n* Maintaining one codebase for pandas, Spark, Dask and Ray projects. Logic and execution are decoupled through Fugue, enabling users to be focused on their business logic rather than writing framework-specific code.\n* Improving iteration speed of big data projects. Fugue seamlessly scales execution to big data after local development and testing. By removing PySpark code, unit tests can be written in Python or pandas and ran locally without spinning up a cluster.\n\n\n\nThe best way to get started with Fugue is to work through the 10 minute tutorials:\n\n* [Fugue in 10 minutes](https://fugue-tutorials.readthedocs.io/tutorials/quick_look/ten_minutes.html)\n* [FugueSQL in 10 minutes](https://fugue-tutorials.readthedocs.io/tutorials/quick_look/ten_minutes_sql.html)\n\n### Python setup\n\nRequires an installation of Python 3.8+.\n\nWe recommend using a Python virtual environment manager such as pipenv, conda or virtualenv.\n\n### Installation\n\nInstall `prefect-fugue` with `pip`:\n\n```bash\npip install prefect-fugue\n```\n\nIt's also recommended to register Fugue blocks into your current Prefect workspace:\n\n```bash\nprefect block register -m prefect_fugue\n```\n\n### Creating a Block\n\nThis will allow a creation of the Fugue Engine block\n\n\n\nThere are 4 items that need to be filled to create a block.\n\n* Block Name - name that will be used to use the block.\n* Engine Name - one of the Fugue supported backends (spark, dask, ray, duckdb)\n* Engine Config - configurations related to the cluster\n* Secret Config - credentials to connect to a cluster\n\nFor example, a Databricks Block could look like:\n\n* Block Name - databricks\n* Engine Name - spark\n* Engine Config - None\n* Secret Config - seen below\n\n```json\n{\n \"host\": \"https://dbc-38aaa459-faaf.cloud.databricks.com\",\n \"token\": \"dapiecaaae64a727498daaaaafe1bace968a\",\n \"cluster_id\": \"0612-191111-6fopaaaa\"\n}\n```\n\nAs long as you installed `prefect_fugue`, Fugue is able to recognize and convert a block expression to a `FugueExecutionEngine`. For example if you have a block with path `fugue/databricks`, then the expression `prefect:fugue/databricks` becomes a valid execution engine expression. When fugue parse this expression, it will load the `Block` from `fugue/databricks`, then base on the fields of the block, it will construct a `DatabricksExecutionEngine` for your Fugue operations.\n\n### Using a Spark Cluster Inside a Flow\n\nLet\u2019s start by running code on top of Databricks. `databricks-connect` is already installed in this environment. This section may have a lot of logs because of the monitoring provided by Prefect. This section also assumes that the user has Prefect configured to the right workspace.\n\nBelow we have one task that takes in a `SparkSession` and uses it to run some Spark code. We can then use this in the Prefect Flow with the `fugue.api.engine_context`. This will create an ephemeral cluster to run the code underneath, and then turn off when finished.\n\n```python\nfrom prefect import task, flow\nimport fugue.api as fa\n\n@task\ndef my_spark_task(spark, n=1):\n df = spark.createDataFrame([[f\"hello spark {n}\"]], \"a string\")\n df.show()\n\n@flow\ndef spark_flow(engine):\n with fa.engine_context(engine) as spark_engine:\n my_spark_task(spark_engine.spark_session, 1)\n\nspark_flow(\"prefect:fugue/databricks\") # pay attention to the engine expression\n```\n\nSimilarly, if you don\u2019t use Databricks but have your own way to get a `SparkSession`, you can directly pass the `SparkSession` into the Flow.\n\n```python\nfrom pyspark.sql import SparkSession\nspark = SparkSession.builder.getOrCreate()\n\nspark_flow(spark)\n```\n\n## More Flexibility\n\n`fugue.api.engine_context` creates a context under which all fugue operations will use the context engine by default. This works either the fugue code is directly under the context or inside the tasks or flows that are invoked under this context.\n\n```python\nfrom prefect import task, flow\nimport fugue.api as fa\n\ndef my_transformer(df:pd.DataFrame) -> pd.DataFrame:\n return df\n\n@task\ndef sub_task(path:str):\n df = fa.load(path)\n df = fa.transform(df, my_transformer, schema=\"*\")\n df.save(path+\".output.parquet\")\n\n@flow\ndef sub_flow(path:str):\n df = fa.load(path)\n df = fa.transform(df, my_transformer, schema=\"*\")\n df.save(path+\".output.parquet\")\n\n@flow\ndef main_flow(path, engine=None):\n with fa.engine_context(engine) as spark_engine:\n sub_task(path)\n sub_flow(path)\n\nmain_flow(\"<local path>\") # test locally, all tasks and flows run without Spark\nmain_flow(\"<dbfs path>\", \"prefect:fugue/databricks\") # all tasks and flows will run on Databrickes\n```\n\n## Testing Locally Before Running Map Jobs on Spark, Dask, and Ray \n\nWe showed how to run Spark code on top of a Spark cluster, but the strength of Fugue is decoupling from distributed framework code such as Spark and Dask. Decoupling from these frameworks allows us to test code locally before scaling out to a cluster. In the example below, we simulate having a pandas DataFrame where each row is a job.\n\nWhen testing the Flow, we can pass `None` as the engine so everything runs on Pandas. When ready to scale out, we can pass in our `Block` or `SparkSession`. Fugue\u2019s `transform()` task will use the engine provided by the `fugue_engine` context.\n\n```python\nfrom time import sleep\nimport pandas as pd\nimport fugue.api as fa\nfrom prefect import task, flow\n\n@task\ndef create_jobs(n) -> pd.DataFrame:\n return pd.DataFrame(dict(jobs=range(n)))\n\n# schema: *,batch_size:str\ndef run_one_job(df:pd.DataFrame) -> pd.DataFrame:\n sleep(len(df)*5)\n return df.assign(batch_size=len(df))\n\n@flow\ndef run_all_jobs(n, engine=None):\n jobs = create_jobs(n)\n with fa.engine_context(engine):\n return transform(jobs, run_one_job, partition=\"per_row\", as_local=True)\n```\n\nWe can test the Flow above on a local machine without Spark. We run on one job first.\n\n```python\nrun_all_jobs(1) # run locally on Pandas\n```\n\nBecasue it succeeded, we can now attach our Fugue Databricks `Block` to run on Databricks. Now we run on 8 jobs, and we\u2019ll see that parallelization from the Spark cluster will make this Flow execute faster.\n\n```python\nrun_all_jobs(8, \"prefect:fugue/databricks\") # run on databricks\n```\n\nThere is still some overhead with sending the work, but the time is decreased compared to the expected execution time if ran sequentially (40 seconds).\n\nWe can also use local Dask by passing the string `\"dask\"`. We can also pass a `Dask Client()` or use the Fugue Engine `Block` with [Coiled](https://coiled.io/). More information can be found in the [Coiled cloudprovider docs](https://fugue-tutorials.readthedocs.io/tutorials/integrations/cloudproviders/coiled.html).\n\n```python\nrun_all_jobs(4, dask_client)\n```\n\n## Running SQL on any Spark, Dask, and Duckdb\n\nPrototyping locally, and then running the full job on the cluster is also possible with [FugueSQL](https://fugue-tutorials.readthedocs.io/tutorials/quick_look/ten_minutes_sql.html). DuckDB is a good engine to run SQL queries on flat files or Pandas DataFrames. When ready, we can bring it to SparkSQL on the cluster. Similar to the transform() task shown above, there is also an fsql() task.\n\nHere we can load in data and perform a query with FugueSQL. FugueSQL has additional keywords such as LOAD and SAVE so we can run everything from loading, processing, and saving all on DuckDB or SparkSQL. More information on FugueSQL can be found in the FugueSQL tutorials.\n\n```python\nimport fugue.api as fa\n\n@flow\ndef run_sql(top, engine):\n with fa.engine_context(engine):\n fa.fugue_sql_flow(\"\"\"\n df = LOAD \"https://d37ci6vzurychx.cloudfront.net/trip-data/green_tripdata_2022-01.parquet\"\n\n SELECT PULocationID, COUNT(*) AS ct FROM df\n GROUP BY 1 ORDER BY 2 DESC LIMIT {{top}}\n PRINT\n \"\"\", top=top).run()\n```\n\nTo debug locally without SparkSQL, we can use DuckDB as the engine.\n\n```python\nrun_sql(2, \"duckdb\"); # debug/develop without spark\n```\n\nAgain to run on the cluster, we can use the Databricks Block.\n\n```python\nrun_sql(10, \"prefect:fugue/databricks\")\n```\n\n## Resources\n\nIf you encounter any bugs while using `prefect-fugue`, feel free to open an issue in the [prefect-fugue](https://github.com/fugue-project/prefect-fugue) repository.\n\nIf you have any questions or issues while using `prefect-fugue`, you can find help in the [Fugue Slack community](http://slack.fugue.ai).\n\n## Development\n\nIf you'd like to install a version of `prefect-fugue` for development, clone the repository and perform an editable install with `pip`:\n\n```bash\ngit clone https://github.com/fugue-project/prefect-fugue.git\n\ncd prefect-fugue/\n\npip install -e \".[dev]\"\n\n# Install linting pre-commit hooks\npre-commit install\n```\n",
"bugtrack_url": null,
"license": "Apache License 2.0",
"summary": "Fugue Prefect integration",
"version": "0.0.3",
"project_urls": {
"Homepage": "https://github.com/fugue-project/prefect-fugue"
},
"split_keywords": [
"prefect"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "246dbe5e155e8bd9cf2f7e2f35e517c9cc50f5b6455b5fc7bdf3ae1ef0d7a231",
"md5": "b3b08b24031b628bbb9e4b7269c5b5ce",
"sha256": "0480aba03c9a182af7314efb1f67630b078e3be5b53757dc7898d1c3e4d23da6"
},
"downloads": -1,
"filename": "prefect_fugue-0.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b3b08b24031b628bbb9e4b7269c5b5ce",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 12522,
"upload_time": "2023-10-18T05:28:33",
"upload_time_iso_8601": "2023-10-18T05:28:33.789118Z",
"url": "https://files.pythonhosted.org/packages/24/6d/be5e155e8bd9cf2f7e2f35e517c9cc50f5b6455b5fc7bdf3ae1ef0d7a231/prefect_fugue-0.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "488e784d45cc3c86b947c0b6a199799036c461025d5015de566c99762b77375c",
"md5": "2717a8ab5cf92ef00351b8db7d218932",
"sha256": "b7d005d94aeb50907a0e39372947c7344a08d2e9e1bf37ca6c9aa835d2934659"
},
"downloads": -1,
"filename": "prefect-fugue-0.0.3.tar.gz",
"has_sig": false,
"md5_digest": "2717a8ab5cf92ef00351b8db7d218932",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 30290,
"upload_time": "2023-10-18T05:28:35",
"upload_time_iso_8601": "2023-10-18T05:28:35.404762Z",
"url": "https://files.pythonhosted.org/packages/48/8e/784d45cc3c86b947c0b6a199799036c461025d5015de566c99762b77375c/prefect-fugue-0.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-10-18 05:28:35",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "fugue-project",
"github_project": "prefect-fugue",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "prefect-fugue"
}
JSON
