https://github.com/stitchfix/hamilton logo
Join Slack
Powered by
How can I skip to the next iteration of `Collect` ...
# hamilton-help
s
How can I skip to the next iteration of 
Collect
should any of these 
_filter*
step
's return an empty 
DataFrame
? I know 
Collect
executes as 
async
so iteration may be the wrong terminology, but I'd like the downstream nodes of the empty result to not have to try to do anything with that empty 
DataFrame
to produce the requested node. If its empty then don't fail the other tasks.
Copy code
@extract_fields({
    "child_process_run_id": int,
    "child_process_completed_at": pd.Timestamp,
})
@pipe_input(
    step(_filter_run_bom).when(last_run="bom"),
    step(_filter_run_latest).when(last_run="last"),
    step(_filter_run_date, completion_date=source("completion_date")).when(last_run="completion_date"),
)
def child_process_run_info(child_process_runs: pd.DataFrame) -> Dict[int, pd.Timestamp]:
e
Is this within the parallelizable/collect block?
s
Yes
e
Got it. There’s currently no great way to short-circuit, but what you’d typically do is filter outside so collect doesn’t end up getting run for those that you don’t want it to run for. That said, the cleaner way might just be to ensure downstream items can handle empty datasets…
s
Copy code
class NoRunsFoundError(Exception):
    ...


@extract_fields({
    "child_process_run_id": int,
    "child_process_completed_at": pd.Timestamp,
})
@pipe_input(
    step(_filter_run_bom).when(last_run="bom"),
    step(_filter_run_latest).when(last_run="last"),
    step(_filter_run_date, completion_date=source("completion_date")).when(last_run="completion_date"),
)
def child_process_run_info(
    child_process_runs: pd.DataFrame, 
    child_rec_process_code: str, 
    completion_date: Optional[str]
) -> Dict[int, pd.Timestamp]:
    """Return run id for process from `last_run` logic."""

    try:
        info = {
            "child_process_run_id": child_process_runs["id"].iloc[0],
            "child_process_completed_at": pd.to_datetime(child_process_runs["completed_at"].iloc[0]),
        }
        return info 
    except Exception as e:
        raise NoRunsFoundError(f"No data for process {child_rec_process_code} on date {completion_date}") from e
Could this work with the 
GracefulErrorAdapter
?
e
Ahh, yes! That’s exactly the use-case that was intended.
There’s special stuff for parallel in there
Should be able to just get hte right sentinel value (
None
) from the parallel block. Haven’t run it in a bit but it should be doable (report back)?
s
Yes I'll try it now and revert
👍 1
e
good thinking
s
OK, so it works until the `Collect`:
Copy code
line 140, in download_run_results__default
    for endpoint, completed_at in all_last_run_result_inputs:
        ^^^^^^^^^^^^^^^^^^^^^^
TypeError: cannot unpack non-iterable NoneType object
e
So yeah, the problem is it is set to 
None
. Should be simple enough to iterate through then onlyunpack if it’s not 
None
Copy code
for item in all_last_run_result_inputs:
    if item is None:
        continue
    endpoint, completed_at = item
    ...
s
for endpoint, completed_at in filter(None,all_last_run_result_inputs):
This should do it no?
e
yep (I didn’t actually know 
filter
could take in 
None
, but turns out it can…)
s
beautiful thank you sir, that was it! and thanks to whoever thought up/made the 
GracefulErrorAdapter
🙌 1
e
Great! Yes, I’m pretty excited about it. I had the idea but James Arruda made it work with Parallel
s
Can I have a step with multiple steps?
Copy code
@pipe_input(
    step(_filter_out_noise).when_in(cond=["a", "b"]), 
    step(_filter_out_more_stuff).when_in(cond=["a", "b"]), 
    step(_filter_out_noise).when_in(cond=["c", "d"]),
)
def clean_data(raw_data: pd.DataFrame) -> pd.DataFrame:
    return raw_data
Essentially, the above probably work, but I am checking to see if 
group
could work here?
Copy code
@pipe_input(
    group(step(_filter_out_noise), step(_filter_out_more_stuff)).when_in(cond=["a", "b"]), 
    step(_filter_out_noise).when_in(cond=["c", "d"]),
)
def clean_data(raw_data: pd.DataFrame) -> pd.DataFrame:
    return raw_data
And of course, the classic 
config.when_in
would work too.
e
This is not supported (cool idea though… Group is a different algebra. I get that it’s a bit more succinct — that said, is there another reason for grouping them?
s
No reason beyond really enjoying the 
step().when_in()
api that enables multiple scenarios on the same node.
🎉 1
The first code block work, no?
e
Yeah, so 
when_in
should work — I see, you want them as the same criteria
Makes sense
s
Without, grouping these into their own defined function by scenario could this work? But then, I'd probably lose debugability...
Copy code
step(lambda _df: _df.pipe(_filter_out_noise).pipe(_filter_out_netted_updowns)).when_in(cond=["a", "b"])
e
Yep! That would be valid. And yeah - trade-off around visibility. @Jernj Frank might have more insight into this
j
have you tried stacking two 
pipe_input
and setting the global 
when_in
for each?
never worked with lambda functions sorry, but can look into it / think about how something like 
group
would look like
e
What would the global 
when_in
look like? Oh, I see, decorating with 
@config.when_in
?
j
yeah, but it is a long shot
I suspect it won't work, but that sound like an interesting feature in itself to be able to turn you hamilton decorators on and off a. runtime
e
Hmm, what would the code look like? And yeah, agreed, definitely an interesting one
j
Good question, it could work if we mark it in the underlying function and the short circuit the node lifecycle cases
s
From the docs example 3, where does this 
upstream_int
come from?
Copy code
@pipe_input(
    step(_add_one)
    step(_add_two, y=source("upstream_node")),
    on_input = "p2"
)
def final_result(p1: int, p2: int, p3: int) -> int:
    return upstream_int
e
Nowhere, heh. That appears broken.
I think we just want 
p1 + p2 + p3
or something like that. If you want to fix that would be appreciated 🙂
s
Ok, I was trying to think of some way with 
named
and 
on_input
when I stumbled across that. I think we already proposed 2 good ways so I'm happy. How can I go about amending? When I hit the edit from the Hamilton Docs page, it sends me to the 
rst
in github
e
Ahh yep, so it’s a bit of a layer of abstraction. See this file for where it goes (it automatically generates the docs…). Should be a one-line change. In fact, it’s correct in the tests.
Open in Slack
PreviousNext
Powered by