Kedro Dynamische Pipelines

• Implementierung von "Kern"-ML-Pipelines, die für verschiedene Geschäftsanwendungen konfiguriert und wiederverwendet werden können.
  Beispielhafte Industrieanwendungen:
  • Einzelhandel: Dies kann den Aufbau eines Umsatzprognosemodells pro Geschäft oder pro Produkt oder die Erstellung personalisierter Empfehlungssysteme für verschiedene Kundensegmente beinhalten.
  • Finanzen: Dies könnte die Entwicklung von Risikobewertungsmodellen für verschiedene Arten von Darlehen oder Kreditprodukten beinhalten.
• Automatische Durchführung mehrerer Experimente, um zu ermitteln, welche Modellkonfiguration am besten abschneidet. Diese Experimente können sich in Bezug auf die verwendeten Funktionen, Modellparameter oder sogar die Art der verwendeten Modelle unterscheiden.

def create_pipeline(**kwargs) -> Pipeline:
    data_science_pipeline = pipeline(
        [
            node(
                func=split_data,
                inputs=["model_input_table", "params:model_options"],
                outputs=["X_train", "X_test", "y_train", "y_test"],
                name="split_data_node",
            ),
            node(
                func=train_model,
                inputs=["X_train", "y_train"],
                outputs="regressor",
                name="train_model_node",
            ),
            node(
                func=evaluate_model,
                inputs=["regressor", "X_test", "y_test"],
                outputs=None,
                name="evaluate_model_node",
            ),
        ]
    )
    baseline = pipeline(
        data_science_pipeline,
        parameters={"params:model_options": "params:model_options"},
        inputs={"model_input_table": "model_input_table",},
        namespace="baseline",
    )
    candidate = pipeline(
        data_science_pipeline,
        inputs={"model_input_table": "model_input_table",},
        tags=["candidate"],
        namespace="candidate"
    )
    return baseline + candidate

model_options:
  test_size: 0.2
  random_state: 3
  features:
    - engines
    - passenger_capacity
    - crew
    - d_check_complete
    - moon_clearance_complete
    - iata_approved
    - company_rating
    - review_scores_rating

candidate:
  model_options:
    test_size: 0.2
    random_state: 666
    features:
      - engines
      - passenger_capacity
      - crew
      - d_check_complete
      - moon_clearance_complete

preprocessed_companies:
  type: pandas.ParquetDataSet
  filepath: data/02_intermediate/preprocessed_companies.pq

preprocessed_shuttles:
  type: pandas.ParquetDataSet
  filepath: data/02_intermediate/preprocessed_shuttles.pq

model_input_table:
  type: pandas.ParquetDataSet
  filepath: data/03_primary/model_input_table.pq

baseline.preprocessed_companies:
  type: pandas.ParquetDataSet
  filepath: data/02_intermediate/preprocessed_companies.pq

baseline.preprocessed_shuttles:
  type: pandas.ParquetDataSet
  filepath: data/02_intermediate/preprocessed_shuttles.pq

baseline.model_input_table:
  type: pandas.ParquetDataSet
  filepath: data/03_primary/model_input_table.pq

candidate.preprocessed_companies:
  type: pandas.ParquetDataSet
  filepath: data/02_intermediate/candidate/preprocessed_companies.pq

candidate.preprocessed_shuttles:
  type: pandas.ParquetDataSet
  filepath: data/02_intermediate/candidate/preprocessed_shuttles.pq

candidate.model_input_table:
  type: pandas.ParquetDataSet
  filepath: data/03_primary/candidate/model_input_table.pq

• Anwendungsfall 1: Sie haben eine Pipeline, die Sie auf einem Datensatz ausführen möchten, der sich im Laufe der Zeit weiterentwickelt - z.B. ein Prognosemodell mit monatlichen Datentabellen, bei dem ein Monat die Daten des Vormonats verbraucht usw., wie in diesem Beispiel.
• Anwendungsfall 2: Sie haben eine Reihe von ähnlichen Modellexperimenten mit ähnlichen Parametern, die Sie parallel durchführen möchten. Die Modellparameter, die verwendeten Features, die Zielspalten oder die Modelltypen können in den verschiedenen Experimenten variieren.
• Anwendungsfall 3: Sie möchten eine "zentrale" / "wiederverwendbare" Pipeline implementieren, die für mehrere geschäftliche Anwendungsfälle konfiguriert und mehrfach ausgeführt werden kann.

• OmegaConfigLoader mit benutzerdefinierten Auflösern
• Dataset-Fabriken
• Modulare Pipelines mit Namespaces
• Zentralisiert settings.py

from kedro.config import OmegaConfigLoader
CONFIG_LOADER_CLASS = OmegaConfigLoader

from copy import deepcopy
import omegaconf

def merge_dicts(dict1, dict2):
    """
    Recursively merge two dictionaries.

    Args:
        dict1 (dict): The first dictionary to merge.
        dict2 (dict): The second dictionary to merge.

    Returns:
        dict: The merged dictionary.
    """
    result = deepcopy(dict1)
    for key, value in dict2.items():
        if (
            key in result
            and isinstance(result[key], omegaconf.dictconfig.DictConfig)
            and isinstance(value, omegaconf.dictconfig.DictConfig)
        ):
            result[key] = merge_dicts(result[key], value)
        else:
            result[key] = value
    return result

DYNAMIC_PIPELINES_MAPPING = {
    "reviews_predictor": ["base", "test1"],
    "price_predictor": ["base", "candidate1", "candidate2", "candidate3"],
}

# The model_options below are "base" options for all pipelines

model_options:
  test_size: 0.2
  random_state: 3
  target: costs
  features:
    - engines
    - passenger_capacity
    - crew
    - d_check_complete
    - moon_clearance_complete
    - iata_approved
    - company_rating
    - review_scores_rating
  model: sklearn.linear_model.LinearRegression
  model_params: {}

# -------------------------------
# Pipeline-specific configuration

price_predictor:
  _overrides:
    target: price
  model_options: ${merge:${model_options},${._overrides}}

  base:
    model_options: ${..model_options}
  
  candidate1:
    _overrides:
      features:
      - engines
      - passenger_capacity
      - crew
      - d_check_complete
      - company_rating
    model_options: ${merge:${..model_options},${._overrides}}

  candidate2:
    _overrides:
      model_params:
        gamma: 2.5
    model_options: ${merge:${..model_options},${._overrides}}

  candidate3:
    _overrides:
      model: sklearn.ensemble.RandomForestRegressor
      model_params:
        max_depth: 1.0

    model_options: ${merge:${..model_options},${._overrides}}

<namespace>:
    <variant - e.g. model variant / experiment name>:
        _overrides: {} # what to override from root configuration
        
	# any specific parameter, e.g. model_options, used in Kedro pipeline
       model_options: ${merge:${..model_options},${._overrides}}

"reviews_predictor": ["base", "test1"],
 "price_predictor": ["base", "candidate1", "candidate2", "candidate3"],

price_predictor.base.regressor:
  type: pickle.PickleDataSet
  filepath: data/06_models/price_predictor/base/regressor.pickle
  versioned: true

price_predictor.candidate1.regressor:
  type: pickle.PickleDataSet
  filepath: data/06_models/price_predictor/candidate1/regressor.pickle
  versioned: true

price_predictor.candidate2.regressor:
  type: pickle.PickleDataSet
  filepath: data/06_models/price_predictor/candidate2/regressor.pickle
  versioned: true

# ... and 9 more entries

"{namespace}.{variant}.regressor":
  type: pickle.PickleDataSet
  filepath: data/06_models/{namespace}/{variant}/regressor.pickle
  versioned: true

"{namespace}.model_input_table":
  type: pandas.ParquetDataSet
  filepath: data/03_primary/{namespace}/model_input_table.pq

from <project_name> import settings

def create_pipeline(**kwargs) -> Pipeline:
    data_processing = pipeline(
        [
            node(
                func=preprocess_companies,
                inputs="companies",
                outputs="preprocessed_companies",
                name="preprocess_companies_node",
            ),
            node(
                func=preprocess_shuttles,
                inputs="shuttles",
                outputs="preprocessed_shuttles",
                name="preprocess_shuttles_node",
            ),
            node(
                func=create_model_input_table,
                inputs=["preprocessed_shuttles", "preprocessed_companies", "reviews"],
                outputs="model_input_table",
                name="create_model_input_table_node",
            ),
        ]
    )

    pipes = []
    for namespace in settings.DYNAMIC_PIPELINES_MAPPING.keys():
        pipes.append(
            pipeline(
                data_processing,
                inputs={
                    "companies": "companies",
                    "shuttles": "shuttles",
                    "reviews": "reviews",
                },
                namespace=namespace,
                tags=settings.DYNAMIC_PIPELINES_MAPPING[namespace],
            )
        )
    return sum(pipes)

from <project name> import settings

def create_pipeline(**kwargs) -> Pipeline:
    data_science_pipeline = pipeline(
        [
            node(
                func=verbose_params,
                inputs=["params:model_options"],
                outputs=None,
                name="debug_node",
            ),
            node(
                func=split_data,
                inputs=["model_input_table", "params:model_options"],
                outputs=["X_train", "X_test", "y_train", "y_test"],
                name="split_data_node",
            ),
            node(
                func=train_model,
                inputs=["X_train", "y_train"],
                outputs="regressor",
                name="train_model_node",
            ),
            node(
                func=evaluate_model,
                inputs=["regressor", "X_test", "y_test"],
                outputs=None,
                name="evaluate_model_node",
            ),
        ]
    )
    pipes = []
    for namespace, variants in settings.DYNAMIC_PIPELINES_MAPPING.items():
        for variant in variants:
            pipes.append(
                pipeline(
                    data_science_pipeline,
                    inputs={"model_input_table": f"{namespace}.model_input_table"},
                    namespace=f"{namespace}.{variant}",
                    tags=[variant, namespace],
                )
            )
    return sum(pipes)

• Führen Sie alles für einen der Namensräume aus, z.B. kedro run --namespace price_predictor: Dies führt die data_processing Pipeline einmal aus und alle Varianten der data_sciencePipeline: base, candidate1, candidate2 usw.
• Führen Sie nur die eine Variante aus: kedro run --namespace price_predictor.candidate1 führt nur die data_science Pipeline in der candidate1 Variante aus und überspringt den Rest. Dazu muss die data_processing Pipeline zuerst ausgeführt werden, um Eingabedaten zu verarbeiten.
• Führen Sie nur die einzelne Variante aus und lassen Sie auch die data_processing Pipeline laufen: kedro run --namespace price_predictor --tags candidate1

• Vorschlag für Pipeline-Filterungshaken, die die Erstellung von Pipelines auf der Grundlage von Parametern/Konfiguration ermöglichen https://github.com/kedro-org/kedro/issues/3000
• Konsistenz der Initialisierung des Config Loaders https://github.com/kedro-org/kedro/issues/3093
• Möglichkeit, mehrere Namespaces gleichzeitig auszuführen https://github.com/kedro-org/kedro/issues/3056
• Hinzufügen einer Validierung - in unseren Projekten, die dynamische Pipelines nutzen, haben wir auch einen Validierungsschritt hinzugefügt, der eine ordnungsgemäße Struktur auf parameters.yml in Bezug auf die definierten Namespaces und Varianten gewährleistet. Wenn Sie daran oder an anderen ML / MLOps Best Practices und Implementierungen interessiert sind, sehen Sie sich unsere MLOps-Demo an und melden Sie sich für eine kostenlose Beratung bei mir an.