Feature/kfp support for lightweight components

[GeorgesLorre]

Tested successfully on vertex

TODO:

• test on kfp
• add compile tests that have lightweight components
• investigate duplicate data dir's

[GeorgesLorre]

FYI the pipeline I ran on vertex:

import dask.dataframe as dd
import pandas as pd
import pyarrow as pa
from fondant.component import DaskLoadComponent, PandasTransformComponent
from fondant.pipeline import Pipeline, lightweight_component

pipeline = Pipeline(
    name="lightweight-pipeline",
    base_path="./data",
)


@lightweight_component(
    base_image="python:3.10",
    extra_requires=[
        "fondant[component,aws,azure,gcp]@git+https://github.com/ml6team/fondant@main"
    ],
)
class CreateData(DaskLoadComponent):
    def load(self) -> dd.DataFrame:
        df = pd.DataFrame(
            {
                "x": [1, 2, 3],
                "y": [4, 5, 6],
            },
            index=pd.Index(["a", "b", "c"], name="id"),
        )
        return dd.from_pandas(df, npartitions=1)


dataset = pipeline.read(
    ref=CreateData,
    produces={"x": pa.int32(), "y": pa.int32()},
)


@lightweight_component(
    base_image="python:3.10",
    extra_requires=[
        "fondant[component,aws,azure,gcp]@git+https://github.com/ml6team/fondant@main"
    ],
)
class AddN(PandasTransformComponent):
    def __init__(self, n: int = 1, **kwargs):
        self.n = n

    def transform(self, dataframe: pd.DataFrame) -> pd.DataFrame:
        dataframe["x"] = dataframe["x"].map(lambda x: x + self.n)
        return dataframe


_ = dataset.apply(
    ref=AddN,
    produces={"x": pa.int32(), "y": pa.int32()},
    consumes={"x": pa.int32(), "y": pa.int32()},
)


if __name__ == "__main__":
    from fondant.pipeline.runner import VertexRunner

    pipeline.base_path = "REDACTED"

    runner = VertexRunner(
        project_id="REDACTED",
        region="europe-west1",
        service_account="REDACTED",
    )
    runner.run(input=pipeline)

[GeorgesLorre]

I'm contemplating removing the KubeflowComponentSpec from the component_spec.py and integrating it more in the compiler (the only place where it is used)

[PhilippeMoussalli]

yes please

[PhilippeMoussalli]

Thanks Geroges!

Are the ToDos still to be tackled in this PR?

[PhilippeMoussalli]

yes please

[GeorgesLorre]

Thanks Geroges!

Are the ToDos still to be tackled in this PR?

Yes I will do them

[PhilippeMoussalli]

Thanks Georges!

[mrchtr]

Lgtm!