This message was deleted.

You have a few options — it depends a bit on how many files there are. Specifically, either: 1. Its the same # of files every time (as you said, taht set represetns that same set of things) 2. Its a different # of files every time (same set of things, different number) For (1), you’re exactly right —

@parameterized_subdag

allows you to process each one individually. Something like:

dr = driver.Driver(*modules, config={})
# the following is a *fixed* set of files
result = dr.execute(
    ..., 
    inputs={
        'file_1_path' : 'path_for_file_1', 
        'file_2_path' : 'path_for_file_2', 
        'file_3_path' : 'path_for_file_3', 
         ...
    }
)

Then, in your

parameterized_subdag

, you can refer to each one specifically:

@parameterized_subdag(
    file_processing_module,
    parameterization={
        "file_1_processed" : {"inputs" : {"path" : source("file_1")}},
        "file_2_processed" : {"inputs" : {"path" : source("file_2)}},
        "file_3_processed" : {"inputs" : {"path" : source("file_3)}},
        ...}
)
def file_n_processed(final_result_from_subdag: pd.DataFrame) -> pd.DataFrame:
    return final_result_from_subdag
    ...

def merged(
    file_1_processed: pd.DataFrame, 
    file_2_processed: pd.DataFrame, 
    file_3_processed: pd.DataFrame, ...) -> pd.DataFrame:
    return ...

@Amos just to chime in. Doing :

values_to_use = load_from_file(...)

@paramaterized_subdag(..., parameterization={... from values_to_use})
def my_subdag(...) -> ...

isn’t unreasonable either. The design choice is really about where do you want the friction for maintaining things? What do you want a pull request to change, and where?

Thanks, guys. This is helpful. I'll give it a go.

Looking — let me see if I can repro

It lives! Nice. Funnily enough, I was pondering that difference last night, but my kids came home to slobber on the keyboard before I got around to investigating further. I'll see if I can prototype the rest of the pipeline and add the parameterisation I want. But that's a good start, thanks.

Of course! For

resolve

, the trick with

resolve

is that the parameter name corresponds to a field in the config dict. E.G:

@resolve(
    when=ResolveAt.CONFIG_AVAILABLE,
    decorate_with=lambda file_names: parameterized_subdag(
        per_file_transforms,
        **{file_name + "_processed": {
            "inputs" : {
                "file_params" : source(file_name), 
                "connector_thinger" : source("connector_thinger")} for file_name in file_names}
    )
)
def ...

Then you’d instantiate a driver with

driver.Driver({'file_names': ['file_1', 'file_2', 'file_3']})

@Amos you’re jumping right into advanced user power mode! To double down on one of Elijah’s questions: is the config checked in, or is that something you expect users to manipulate for input to the code? If it’s checked in — then my suggestion would be to just make it code and save the gymnastics of going from YAML to python. Otherwise, a note on

resolve

you needn’t define the function inline with a lambda - you can pass in a function that does that - just remember to use

_

in front of it if you’re defining it in the same python module it’ll be used in.

Really appreciate the input, guys. Maybe it's helpful if I take half a step back and explain what I'm trying to do. I want a generic pattern for preprocessing common scientific data formats, which include tabular data. Hamilton seems attractive because the trunks of pipelines could be written in its declarative style with Pandera-based validation at each end. The pipelines need to be conceptually segregated into things that happen to specific chunks of data (files/query results), things that happen to specific types of data and things are required for specific models. The latter two can be relatively static, and Hamilton's decorator-based parameterisation seems more than adequate. It's the first stage that's a bit sticky. These pipelines are to be written for experimental applications by people who don't much care about software engineering; they may be run by people with almost no coding experience, and there isn't a large maintenance team. While the data should come from proper storage, in practice it often doesn't. For better or worse, the models end up being more useful when people who turn up with a heap of nasty files can add a simple YAML config that helps them meet the minimum data requirements. Again in a toy example, this is the sort of thing:

# Define a YAML string
yaml_str = """
import:
  type: csv
  file_path: data.csv
transformations:
  - method: dropna
  - method: rename
    args:
      columns:
        old_column_name: new_column_name
"""
...
import pandas as pd
import yaml


class BaseImporter:
    def __init__(self, config):
        self.config = config

    def read_data(self):
        raise NotImplementedError


class CsvImporter(BaseImporter):
    def read_data(self):
        return pd.read_csv(self.config["file_path"])


class Transformer:
    def __init__(self, config, custom_functions={}):
        self.config = config
        self.custom_functions = custom_functions

    def apply_transformations(self, df):
        for transformation in self.config:
            method = transformation["method"]
            args = transformation.get("args", {})

            if method in self.custom_functions:
                func = self.custom_functions[method]
                df = func(df, **args)
            else:
                df = getattr(df, method)(**args)
        return df


def import_and_transform(yaml_config_file):
    with open(yaml_config_file, "r") as file:
        config = yaml.safe_load(file)

    # Create the appropriate importer based on the config
    importer_class = {
        "csv": CsvImporter,
    }[config["import"]["type"]]
    importer = importer_class(config["import"])

    # Read data using the importer
    data = importer.read_data()

    # Apply transformations
    transformer = Transformer(config["transformations"])
    transformed_data = transformer.apply_transformations(data)

    return transformed_data

The point is to dynamically cater for a minimal set of clean-up operations without writing extra code. I'm trying at the moment to see if I can prototype that in Hamilton. In something closer to a real-world example, the YAML might be marginally more complicated. At this stage I can change the structure.

yaml_str = """
config:
    io_config:
        project_key: SOMEPROJ
        folder_name: SOMEFLDR
inputs:
    file_1:
        file_path: some_data.csv
        file_type: csv
        load_args:
            usecols:
            - MYCOLUMN
        transformations:
            etc. 
...

It makes sense to run this first stage in a subDAG. At the moment I instantiate a data connector in the DAG and pass it in, along with the params from the YAML … something like this, as modified from your guidance.

@resolve(
    when=ResolveAt.CONFIG_AVAILABLE,
    decorate_with=lambda inputs: parameterized_subdag(
        per_file_transforms,
        **{
            file_id
            + "_processed": {
                "inputs": {
                    "file_params": file_params,
                    "folder_handle": source("folder_handle"),
                },
            }
            for file_id, file_params in inputs.items()
        },
    ),
)
def read_files(
...

But the

file_params

part is problematic because hamilton wants to call methods on it that suggest it should be specified as an upstream dependency. I guess I need to break up the inputs dict somehow with a node before this one? Ideally, I want three steps in the subDAG: one for loading, one for the clean-up transforms and one for merging the result to something suitable for validation. At some future time when the data comes from real storage, the subDAG stage is then easy to remove. At the highest level, the overarching problem I'm looking to solve is working with really smart people who hide brilliant models behind goopy setup code that means you pretty much have to understand the whole thing before you can apply it to anything else. That's probably too much info. But if I can make this work with Pandas then the next Q becomes can it work with GeoPandas and then with various other things. Better dash as my kids are screaming. Appreciate any tips. And if you think this is the wrong approach or not right for Hamilton, say so. I won't be upset. Cheers…

@Amos thanks for the context. Let’s set up a call to draw out a few things. I think I understand, but I think drawing out a few things would help me grok and provide the right recommendations. 🙂

Pre: call — a few points: 1. re:

file_params

, I think you want a

value

on top of it. E.g.

"file_params" : value(file_params)

Dependencies can be

source

or

value

— source means you get it from an upstream node,

value

means you get it from a literal value. E.G.

source('foo')

means there’s a node

foo

to depend on and

value(3)

means its the literal value

3

. I think that’ll solve your thing. 2. Geopandas is fully supported on hamilton (see https://github.com/DAGWorks-Inc/hamilton/blob/5c8e564d19ff2386b00a1ef71de43d11205c5273/hamilton/plugins/geopandas_extensions.py)

Ah

value(…)

, of course. Why didn't I remember that. That worked, and I was able to complete the subDAG with the

inject(…)

parts etc. You guys have been very generous with your time, and I'm hesitant to take up more of it. But if you're up for a chat, I'm on UTC+9:30 and can be pretty flexible outside child-chaos hours. I'm hoping this exercise becomes an example I can pitch to a wider array of users, so maybe there's something in it for your project in the medium term. I'd certainly gratefully receive any tips.

We have customers in australia so happy to make do (also somehow we’re on all hours of the night 😆 )

That's kind of you. I have a few projects on the go, so perhaps if I keep forging ahead until I get truly stuck or have done the best I can then we could organise a call about how to move forward? In the interim, here's one thing that's confusing me. Why doesn't this work?

@inject(join_order=value(_JOIN_ORDER), merge_cols=value(_MERGE_COLS))
@resolve(
    when=ResolveAt.CONFIG_AVAILABLE,
    decorate_with=lambda inputs: inject(
        curated_inputs=group(**{key: source(key + _CURATED) for key in inputs}),
    ),
)
def joined_files(
    curated_inputs: Dict[str, pd.DataFrame],
    join_order: List[str],
    merge_cols: List[str],
) -> pd.DataFrame:

I get

AttributeError: 'str' object has no attribute '__name__'

In this context,

_JOIN_ORDER

is just a list of strings defined in the same module.

So, I haven’t tested dual injects, but I think that’s a current limitation — in this case it might not be catching it (there’s usually a catch), can do a bit of testing to figure out why. This isn’t an inherent limitation, more an implementation detail. That said, I don’t think

inject

outside of the

resolve

is actually what you want… Rather, hardcoding it in the function, or as a default, might be a lot cleaner (yes, defaults can’t be mutable, but this could be a tuple or a sentinel

None

value…). That way you don’t have to do two hops when reading it.