Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

`MultiProcessingExecutor` pickle'ing error when using `extract_fields`

Aperta
  • Qualcosa non va?
#1,285 2 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 2 giorni

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
40/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Ferma
Stack tecnologico
pandas, python
Ambito
data-engineering

Direzione di ricerca

Inizia da hamilton/execution/executors.py intorno a get_state e dall’implementazione di extract_fields.transform_node indicata nel traceback. Esegui la riproduzione fornita su Windows/Python 3.12 con MultiProcessingExecutor e confrontala con le modalità sincrona e multithreading. Il lavoro è completato quando l’esempio extract_fields termina in multiprocessing senza l’errore di pickle relativo a un oggetto locale.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

triage

Current behavior

MultiProcessingExecutor pickle'ing error when using extract_fields.

Stack Traces

  File "c:\codebase\api\venv\Lib\site-packages\hamilton\execution\executors.py", line 204, in get_state
    self.future.result()
  File "C:\Program Files\Python312\Lib\concurrent\futures\_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python312\Lib\concurrent\futures\_base.py", line 401, in __get_result
    raise self._exception
  File "C:\Program Files\Python312\Lib\multiprocessing\queues.py", line 244, in _feed
    obj = _ForkingPickler.dumps(obj)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Program Files\Python312\Lib\multiprocessing\reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
AttributeError: Can't pickle local object 'extract_fields.transform_node.<locals>.dict_generator'

Screenshots

Image

Steps to replicate behavior

import pandas as pd
from typing import Dict, List, Tuple

from hamilton.htypes import Parallelizable, Collect
from hamilton.function_modifiers import extract_fields


def _load_full_dataset() -> pd.DataFrame:
    url = "https://raw.githubusercontent.com/mwaskom/seaborn-data/master/penguins.csv"
    return pd.read_csv(url)


def kind(kinds: List[str]) -> Parallelizable[str]:
    for kind in kinds:
        yield kind


def raw_kind(kind: str) -> pd.DataFrame:
    return df[df.species == kind]


@extract_fields(dict(raw_df=pd.DataFrame, agg_df=pd.DataFrame))
def all_kinds(raw_kind: Collect[pd.DataFrame]) -> Dict[str, pd.DataFrame]:
    raw = pd.concat(raw_kind)
    return {
        "raw_df": raw,
        "agg_df": raw.groupby("species")[
            ["bill_length_mm", "bill_depth_mm", "flipper_length_mm", "body_mass_g"]
            ].mean().reset_index(),
    }

def all_data(raw_df: pd.DataFrame, agg_df: pd.DataFrame) -> Tuple[pd.DataFrame]:
    return raw_df, agg_df

if __name__ == "__main__":
    from hamilton import driver, settings
    from hamilton.execution import executors

    import __main__
    
    def debug_parallellism(mode: str) -> Tuple:
        if mode == "local":
            executor = executors.SynchronousLocalTaskExecutor()
        elif mode=="multithreading":
            executor = executors.MultiThreadingExecutor(max_tasks=4)
        elif mode=="multiprocessing":
            executor = executors.MultiProcessingExecutor(max_tasks=6)
        
        config = {settings.ENABLE_POWER_USER_MODE: True, "kinds": species}
        dr = (
            driver.Builder()
            .with_modules(__main__)
            .with_config(config)
            .enable_dynamic_execution(allow_experimental_mode=True)
            .with_remote_executor(executor)
            .build()
        )

        dag = dr.visualize_execution(["all_data"], f"parallel-penguins-{mode}-dag.png", bypass_validation=True)
        results = dr.execute(["all_data"])
        return dag, results
    
    
    df = _load_full_dataset()
    species = df.species.unique().tolist()

    modes = ["local", "multithreading", "multiprocessing"]
    for mode in modes:
        try:
            dag, results = debug_parallellism(mode)
        except Exception as e:
            print(f"Error in {mode=}: {e}")

Library & System Information

Microsoft Windows [Version 10.0.22621.4890]
Python 3.12
sf-hamilton==1.87.0

Expected behavior

I was hoping this would work since the SynchronousLocalTaskExecutor works as expected.
I tried MultiThreadingExecutor for my use case but it hangs longer than the sync executor which is why I am trying the MultiProcessingExecutor

Lingua principale
Jupyter Notebook
Stelle
2.6k
Fork
218
Merge medio
2g 13h
PR unite (30g)
18

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/hamilton

Tutte le issue di apache/hamilton

Issue simili

Altre issue su Data Engineering

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.