@Iliya R can’t seem to recreate it.

import pandas as pd
from hamilton.function_modifiers import load_from, source
from hamilton.htypes import Parallelizable, Collect

def code(inputs: list) -> Parallelizable[str]:
   for _code in inputs:
      yield _code

@load_from.csv(path=source("code"))
def data_set(df: pd.DataFrame) -> pd.DataFrame:
   # some processing
   return df

def reduced_df(data_set: Collect[pd.DataFrame]) -> pd.DataFrame:
   return pd.concat(data_set)

def rest_of_my_computation(reduced_df: pd.DataFrame) -> object:
    # do something
    return reduced_df

this works for me:

from hamilton import driver
dr = driver.Builder().with_modules(example).enable_dynamic_execution(allow_experimental_mode=True).build()
dr.execute(["rest_of_my_computation"], inputs={"inputs": ["a", "b", "c"]})

Yeah, it does work with a single symbol

BTW the reason I'm using a

@subdag

is I need the list of symbols for multiple loading processes, and so I need to consume it more than once

That said, taking a step back, you’re running the parallel generator inside a subdag, right?

OK, doctored it up, and this works. Not 100% sure about the use of subdag — using it with parallelizable is often a bit redundant. Note that the biggest difference is that I just created

symbol

as a function. Due to limitations (that I think you mentioned on github)

symbol_generator

is not allowed to be consumed twice. So a dummy (as I think you suggested above/hinted at) is actually a nice way to do it

# data_loading.py
from dataclasses import dataclass
from typing import Iterable, Collection, Any

import pandas as pd

from hamilton.function_modifiers import source, load_from, subdag
from hamilton.htypes import Parallelizable, Collect
from hamilton.io.data_adapters import DataLoader


def symbol_generator(symbols: Iterable[str]) -> Parallelizable[str]:
    yield from symbols


@dataclass(frozen=True)
class CustomLoader(DataLoader):
    """Loads data using the API."""
    symbol: str

    # other params

    def load_data(self, _: type[type]) -> tuple[pd.DataFrame, dict[str, Any]]:
        raw_df: pd.DataFrame = pd.DataFrame({"timestamp": [1, 2, 3], "value": [1, 2, 3]})

        return raw_df, {}

    @classmethod
    def applicable_types(cls) -> Collection[type]:
        return [pd.DataFrame]

    @classmethod
    def name(cls) -> str:
        return "mdq"


from hamilton.registry import register_adapter

register_adapter(CustomLoader)


def symbol(symbol_generator: str) -> str:
    return symbol_generator


@load_from.mdq(
    symbol=source("symbol"),
    # other parameters
)
def data_type_q(input_data: pd.DataFrame) -> pd.DataFrame:
    return input_data

def tagged_q(data_type_q: pd.DataFrame, symbol: str) -> pd.DataFrame:
    return data_type_q.assign(symbol=symbol)


def mdq_all_symbols(tagged_q: Collect[pd.DataFrame]) -> pd.DataFrame:
    mdq_df: pd.DataFrame = pd.concat(tagged_q, axis=0, sort=True, ignore_index=True)
    return mdq_df.set_index(["timestamp", "symbol"]).sort_index()


if __name__ == "__main__":
    import __main__
    from hamilton import driver

    dr = driver.Builder().with_modules(__main__).enable_dynamic_execution(allow_experimental_mode=True).build()
    dr.display_all_functions("./output.png")
    dr.execute(["mdq_all_symbols"], inputs={"symbols": ["a", "b", "c"]})

output.png

@Iliya R this thread is related to https://hamilton-opensource.slack.com/archives/C03M33QB4M8/p1724274567358299?thread_ts=1724252682.574579&cid=C03M33QB4M8 right?

Which brings me to the point of nested parallelization... If it's impossible, maybe it's good that it didn't work for me here... I can use it where it matters more

@Elijah Ben Izzy please clarify - did you change something to make it work, or it just worked for you?

Do you have a scheduling tool like the one Stefan does (that shows when you're available)?