<jupyter_start><jupyter_text>Introduction  Utilizing Distributed Computepandas is great for small datasets, but unfortunately does not scale well large datasets. The primary reason is that pandas is single core, and does not take advantage of all available compute resources. A lot of operations also generate [intermediate copies](https://pandas.pydata.org/pandas-docs/stable/user_guide/scale.htmlscaling-to-large-datasets) of data, utilizing more memory than necessary. To effectively handle data with pandas, users preferably need to have [5x to 10x times](https://wesmckinney.com/blog/apache-arrow-pandas-internals/) as much RAM as the size of the dataset.Spark and Dask allow us to split compute jobs across multiple machines. They also can handle datasets that don’t fit into memory by spilling data over to disk in some cases. But ultimately, moving to Spark or Dask still requires significant code changes to port existing pandas code. Added to changing code, there is also a lot of knowledge required to use these frameworks effectively.**Fugue is a framework that is designed to unify the interface between pandas, Spark, and Dask, allowing one codebase to be used across all three engines.**  Fugue `transform()`The simplest way Fugue can be used to scale Pandas based code to Spark or Dask is the `transform()` function. In the example below, we’ll train a model using scikit-learn and pandas, and then perform the inference parallelized on top of the Spark execution engine.<jupyter_code>import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression

X = pd.DataFrame({"x_1": [1, 1, 2, 2], "x_2":[1, 2, 2, 3]})
y = np.dot(X, np.array([1, 2])) + 3
reg = LinearRegression().fit(X, y)<jupyter_output><empty_output><jupyter_text>After training our model, we then wrap it in a `predict()` function. This function is still written in pandas. We can easily test it on the `input_df` that we create. Wrapping it will allow us to bring it to Spark.<jupyter_code>def predict(df: pd.DataFrame, model: LinearRegression) -> pd.DataFrame:
    return df.assign(predicted=model.predict(df))

input_df = pd.DataFrame({"x_1": [3, 4, 6, 6], "x_2":[3, 3, 6, 6]})

# test the function
predict(input_df.copy(), reg)<jupyter_output><empty_output><jupyter_text>Now we bring it to Spark using Fugue using the `transform()` function. This takes in a DataFrame and applies a function to it using either of the Pandas, Spark, or Dask engines. The `transform()` inputs will be explained later, but for now, notice that we did not make modifications to the pandas-based `predict()` function in order to use it on Spark. This function can now scale to big datasets through the Spark execution engine.Even if there is no Spark cluster available, the `SparkExecutionEngine` will start a local Spark instance and parallelize the jobs with all cores of the machine.<jupyter_code>from fugue import transform
import fugue_spark

result = transform(
    input_df,
    predict,
    schema="*,predicted:double",
    params=dict(model=reg),
    engine="spark"
)
print(type(result))
result.show()<jupyter_output><class 'pyspark.sql.dataframe.DataFrame'>
+---+---+---------+
|x_1|x_2|predicted|
+---+---+---------+
|  3|  3|     12.0|
|  4|  3|     13.0|
|  6|  6|     21.0|
|  6|  6|     21.0|
+---+---+---------+<jupyter_text>The `transform()` function takes in the following arguments:* df     - input DataFrame. Can be a Spark, Pandas, or Dask DataFrame* using  - a Python function with valid type annotations* schema - output schema of the operation* params - a dictionary of parameters to pass in the function* engine - the execution engine to run the operation on (Pandas, Spark, or Dask). Because we supplied `"spark"` as the engine, the `predict()` function will be applied on `input_df` on top of the Spark ExecutionEngine. Fugue will handle the conversion from a Pandas DataFrame to a Spark DataFrame. Similarly, a Spark DataFrame can be passed to the `transform()` call. Supplying no engine uses the pandas-based `NativeExecutionEngine`. Fugue also has a `DaskExecutionEngine` available.Explicit `schema` is a hard requirement in distributed computing frameworks, so we need to supply the output `schema` of the operation. When compared to the Spark equivalent (seen below), this is a much simpler interface to handle the `schema`.  ConclusionFugue's `transform()` function can scale pandas-written code to Spark or Dask, without altering the existing functions. In the next section, we’ll see take a deeper look at the `transform()` function. While we used pandas here, we’ll also show that native Python functions can also be used across the different execution engines.  [Optional] Spark Equivalent of `transform()`Below is an example of how the `predict()` function would be brought to Spark without the `transform()` function. This implementation uses the Spark’s `mapInPandas()` method available in Spark 3.0. Note how the `schema` has to be handled inside the `run_predict` function. This is the `schema` requirement we mentioned earlier that Fugue provides a simpler interface for.<jupyter_code>from typing import Iterator, Any, Union
from pyspark.sql.types import StructType, StructField, DoubleType
from pyspark.sql import DataFrame, SparkSession

spark_session = SparkSession.builder.getOrCreate()

def predict_wrapper(dfs: Iterator[pd.DataFrame], model):
    for df in dfs:
        yield predict(df, model)

def run_predict(input_df: Union[DataFrame, pd.DataFrame], model):
    # conversion
    if isinstance(input_df, pd.DataFrame):
        sdf = spark_session.createDataFrame(input_df.copy())
    else:
        sdf = input_df.copy()

    schema = StructType(list(sdf.schema.fields))
    schema.add(StructField("predicted", DoubleType()))
    return sdf.mapInPandas(lambda dfs: predict_wrapper(dfs, model), 
                           schema=schema)

result = run_predict(input_df.copy(), reg)
result.show()<jupyter_output>+---+---+---------+
|x_1|x_2|predicted|
+---+---+---------+
|  3|  3|     12.0|
|  4|  3|     13.0|
|  6|  6|     21.0|
|  6|  6|     21.0|
+---+---+---------+
{'fraction_non_alphanumeric': 0.10124508519003932, 'fraction_numerical': 0.011795543905635648, 'mean_word_length': 4.262931034482759, 'pattern_counts': {'":': 4, '<': 16, '<?xml version=': 0, '>': 17, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '518448', 'license': 'Apache-2.0', 'n_tokens_mistral': 1817, 'n_tokens_neox': 1712, 'n_words': 759, 'path': 'tutorials/beginner/introduction.ipynb', 'repo_name': 'fugue-project/tutorials', 'size': '6104'}