New Pipeline Generation Merged into Main Codebase

[yoonspark]

Overview

We are happy to share that our work on new pipeline generation has been finalized and merged into the main codebase! It is yet to be included in our next release, but you can give it a quick try now if you want:

pip install git+https://github.com/LineaLabs/lineapy.git --upgrade

A major change in the new pipeline generation is that the pipeline's artifact code is now modularized into "non-overlapping" functions, where duplicate code blocks are factored out to reduce redundant computations (which might be expensive to run). For instance, consider the following development code:

import pandas as pd
from sklearn.linear_model import LinearRegression

import lineapy


# Load train data
url1 = "https://raw.githubusercontent.com/LineaLabs/lineapy/main/examples/tutorials/data/iris.csv"
train_df = pd.read_csv(url1)

# Initiate the model
mod = LinearRegression()

# Fit the model
mod.fit(
    X=train_df[["petal.width"]],
    y=train_df["petal.length"],
)

# Save the fitted model as an artifact
lineapy.save(mod, "iris_model")

# Load data to predict (assume it comes from a different source)
pred_df = pd.read_csv(url1)

# Make predictions
petal_length_pred = mod.predict(X=pred_df[["petal.width"]])

# Save the predictions
lineapy.save(petal_length_pred, "iris_petal_length_pred")

Now, say we are building a pipeline with the above artifacts, like so:

lineapy.to_pipeline(
    artifacts=["iris_model", "iris_petal_length_pred"],
    framework="AIRFLOW",
    dependencies={"iris_petal_length_pred": {"iris_model"}},
    pipeline_name="iris",
    output_dir="./iris_pipeline/",
)

With the old version of pipeline generation, we would have gotten the artifact code modularized as the following:

### ./iris_pipeline/iris.py

[...]

def iris_model():
    import pandas as pd
    from sklearn.linear_model import LinearRegression

    url1 = "https://raw.githubusercontent.com/LineaLabs/lineapy/main/examples/tutorials/data/iris.csv"
    train_df = pd.read_csv(url1)
    mod = LinearRegression()
    mod.fit(
        X=train_df[["petal.width"]],
        y=train_df["petal.length"],
    )
    pickle.dump(mod, open("pre-4527270946088797826-post.pkl", "wb"))


def iris_petal_length_pred():
    import pandas as pd
    from sklearn.linear_model import LinearRegression

    url1 = "https://raw.githubusercontent.com/LineaLabs/lineapy/main/examples/tutorials/data/iris.csv"
    train_df = pd.read_csv(url1)
    mod = LinearRegression()
    mod.fit(
        X=train_df[["petal.width"]],
        y=train_df["petal.length"],
    )
    pred_df = pd.read_csv(url1)
    petal_length_pred = mod.predict(X=pred_df[["petal.width"]])
    pickle.dump(petal_length_pred, open("pre-2203554078374588541-post.pkl", "wb"))

[...]

This modularization is not ideal because iris_petal_length_pred() is repeating the same computation as in iris_model(); if this repeated computation were an expensive one, this pipeline would have been extremely inefficient to run.

With the new pipeline generation, we now get the code modularized as the following instead:

### ./iris_pipeline/iris_module.py

[...]

def get_url1_for_artifact_iris_model_and_downstream():
    url1 = "https://raw.githubusercontent.com/LineaLabs/lineapy/main/examples/tutorials/data/iris.csv"
    return url1


def get_iris_model(url1):
    train_df = pd.read_csv(url1)
    mod = LinearRegression()
    mod.fit(
        X=train_df[["petal.width"]],
        y=train_df["petal.length"],
    )
    return mod


def get_iris_petal_length_pred(mod, url1):
    pred_df = pd.read_csv(url1)
    petal_length_pred = mod.predict(X=pred_df[["petal.width"]])
    return petal_length_pred

[...]

As shown, the modularized code now contains "non-overlapping" functions, e.g., get_iris_petal_length_pred() no longer repeats the same computation as that in get_iris_model() (the former instead takes the output from the latter to carry on its own processing). Note also that LineaPy has smartly identified the "common" computation between iris_model and iris_petal_length_pred artifacts, which has been automatically factored out into its own function, i.e., get_url1_for_artifact_iris_model_and_downstream(); if this were a step involving a heavy computation, the current modularization would have saved much time and resource.

In sum, the new pipeline generation modularizes user code to form more efficient pipeline components.

Reference

Related PRs include (listing the latest first):

• LIN-469 Replace existing pipeline file generation #757
• LIN-478 Integrate ArtifactCollection into Pipeline class #750
• LIN-531 Implement PythonOperatorPerArtifact #748
• Implement pipeline writers for artifact collection (includes PR-737) #738
• LIN-468 Support multi-session pipeline building (for new graph refactor) #734
• Lin 467 artifact dependencies within same session #714