Dataframe input to Orchestrator

[JeppeBlixen]

Dataframe input to Orchestrator

[mrmasterplan]

Making use of the fact that objects are handled by reference, here is a simple implementation in pseudocode of your problem:

from typing import Optional
from pyspark.sql import DataFrame
from atc.etl import Transformer, Orchestrator, Loader


class BatchTransformer(Transformer):
    def __init__(self):
        super().__init__()
        self.batch_df: Optional[DataFrame] = None

    def setInputs(self, batch_df):
        self.batch_df = batch_df

    def process(self, df: DataFrame) -> DataFrame:

        df = df.join(self.batch_df)  # etc, your logic

        self.batch_df = None  # we have consumed it
        return df


class StreamLoader(Loader):
    def __init__(self):
        super().__init__()
        self.transformer = BatchTransformer()
        self.orchestrator = (
            Orchestrator()
            # .extract_from(BatchExtractor())
            .transform_with(self.transformer)
            # .load_into(BatchLoader())
        )

    def save(self, df: DataFrame) -> None:
        self.transformer.setInputs(df)
        self.orchestrator.execute()