opendcai/dataflow-webui/pandas-operator

from dataflow.operators.core_text import PandasOperator

PandasOperator(
    process_fn=[
        lambda df: df.rename(columns={"old": "new"}),
        lambda df: df[df["score"] > 0],
    ]
)

op.run(
    storage=self.storage.step(),
)
# returns: empty string ""

from dataflow.operators.core_text import PandasOperator
from dataflow.utils.storage import FileStorage
class MyPipeline:
    def __init__(self):
        self.storage = FileStorage(
            first_entry_file_name="./data/input.jsonl",
            cache_path="./cache",
            file_name_prefix="step",
            cache_type="jsonl"
        )
        self.transformer = PandasOperator(
            process_fn=[
                lambda df: df.assign(score2=df["score"] * 2),
                lambda df: df.sort_values("score", ascending=False),
                lambda df: df.drop(columns=["temp_col"])
            ]
        )
    def forward(self):
        self.transformer.run(
            storage=self.storage.step()
        )
if __name__ == "__main__":
    pipeline = MyPipeline()
    pipeline.forward()