Hi, I found an interesting bug(?maybe):
```from ha...
# hamilton-helpr
Roy Kid
12/09/2024, 10:00 AMHi, I found an interesting bug(?maybe):
Copy code
from hamilton.htypes import Parallelizable, Collect
def stage1(urls: list[str])-> Parallelizable[str]:
for url in urls:
yield url
def substage1(stage1: str) -> str:
return stage1.upper()
def reduce_stage1(substage1: Collect[str]) -> list[str]:
return list(substage1)
def stage2(users: list[str]) -> Parallelizable[str]:
for user in users:
print("user: ", user)
yield user
def substage2(stage2: str, reduce_stage1: list[str]) -> str:
print("stage2: ", stage2)
print("reduce_stage1: ", reduce_stage1)
return stage2
def reduce_stage2(substage2: Collect[str]) -> list[str]:
return list(substage2)KeyError: 'Key reduce_stage2 not found in cache'reduce_stage1reduce_stage1substage2Roy Kid
12/09/2024, 10:03 AMupdate:
Copy code
from hamilton.htypes import Parallelizable, Collect
def stage1(urls: list[str])-> Parallelizable[str]:
for url in urls:
yield url
def substage1(stage1: str) -> str:
return stage1.upper()
def reduce_stage1(substage1: Collect[str]) -> list[str]:
return list(substage1)
def tmp(reduce_stage1: list[str]) -> list[str]:
return reduce_stage1 # ['A', 'B', 'C']
def stage2(users: list[str]) -> Parallelizable[str]:
for user in users:
print("user: ", user)
yield user
def substage2(stage2: str, tmp: list[str]) -> str:
print("stage2: ", stage2) # d, e, f
print("reduce_stage1: ", tmp) # ['A', 'B', 'C']
return stage2
def reduce_stage2(substage2: Collect[str]) -> list[str]:
return list(substage2) Copy code
user: d
user: e
user: f
stage2: ['d', 'e', 'f']
reduce_stage1: ['A']
stage2: ['d', 'e', 'f']
reduce_stage1: ['B']
stage2: ['d', 'e', 'f']
reduce_stage1: ['C']
{'reduce_stage2': [['d', 'e', 'f'], ['d', 'e', 'f'], ['d', 'e', 'f']]}Roy Kid
12/09/2024, 10:04 AMThis one works:
Copy code
from hamilton.htypes import Parallelizable, Collect
def stage1(urls: list[str])-> Parallelizable[str]:
for url in urls:
yield url
def substage1(stage1: str) -> str:
return stage1.upper()
def reduce_stage1(substage1: Collect[str]) -> list[str]:
return list(substage1)
def tmp(reduce_stage1: list[str]) -> list[str]:
return reduce_stage1 # ['A', 'B', 'C']
def stage2(users: list[str], tmp: list[str]) -> Parallelizable[tuple[str, list]]:
for user in users:
print("user: ", user)
yield user, tmp
def substage2(stage2: tuple[str, list]) -> str:
print("stage2: ", stage2[0]) # d, e, f
print("reduce_stage1: ", stage2[1]) # ['A', 'B', 'C']
return stage2
def reduce_stage2(substage2: Collect[str]) -> list[str]:
return list(substage2)e
Elijah Ben Izzy
12/09/2024, 3:31 PMHmm, interesting… Need to dig in more, but seems like a bug? Worth opening up an issue — there are some caveats around parallelism that we’ve been knocking out over time. Tracking some here: https://github.com/DAGWorks-Inc/hamilton/issues/301
r
Roy Kid
12/10/2024, 8:41 AMsorry for reply late... This old issue seems exactly what I encounter here, and should be already fixed. I'm using sf-hamilton 1.83. I move the output of Parallel block from inside to begin of next next Parallel block, and fix it temporarily.
e
Elijah Ben Izzy
12/10/2024, 6:41 PMSo to be clear, not all of those are fixed, but a few of them are — report back when you’ve tried it out? Otherwise if the newest one doesn’t work I’m happy to hepl debug!
r
Roy Kid
12/10/2024, 6:57 PMthx! which version should i use?
e
Elijah Ben Izzy
12/10/2024, 7:00 PMAlways the latest!
sf-hamilton==1.84.0Elijah Ben Izzy
12/10/2024, 7:00 PMBut if you’re using 1.83 then we haven’t fixed it since thne
Elijah Ben Izzy
12/10/2024, 7:00 PMJust to be clear, which of the issues on that umbrella issue do you think this is?
r
Roy Kid
12/10/2024, 7:04 PMsorry, you mean which issue my bug belongs to? i think it should be #1030. am i understand correctly?
e
Elijah Ben Izzy
12/10/2024, 7:05 PMSO in #1030 there are quite a few issues there. I got the sense that you said it would be solved by one of those? If not worth opening a ticket and outlining it with a reproduction (E.G. a full set of code to run to figure out what’s happening). Then I can debug!
r
Roy Kid
12/10/2024, 7:09 PMha! it's simple. the problem is you can not use "result of previous reducer" in "next worker". you should put result in next's mapper, and yield with the value. this can fix the problem.
Roy Kid
12/10/2024, 7:09 PMto reproduce it, you can use my code above
Roy Kid
12/10/2024, 7:09 PMi think the problem is exactly same with 1030
e
Elijah Ben Izzy
12/10/2024, 7:10 PMHmm, that one specifically has the parallelizable directly next to the collect. I don’t think that’s the right one?
r
Roy Kid
12/10/2024, 7:12 PMreally, lets me have a look again with PC
Roy Kid
12/10/2024, 7:17 PMno, they are similar but not mine. sorry for the confusion. my case is more serial than 742, and the first parallel block already finished.
e
Elijah Ben Izzy
12/10/2024, 7:17 PMYep, so open an issue with the code and I’ll dig in
r
Roy Kid
12/10/2024, 7:17 PMalso, i met same error in 1029, but i can't reproduce it. i think maybe because im using notebook