Skip to content

Commit c5f202a

Browse files
lucasvanmollucasvanmol
committed
Rework Dataflow IR
1 parent ace6b52 commit c5f202a

33 files changed

+1229
-833
lines changed

‎deathstar_hotel_reservation/test_demo.py‎

Lines changed: 99 additions & 99 deletions
Original file line numberDiff line numberDiff line change
@@ -1,100 +1,100 @@
11

2-
import os
3-
import sys
4-
5-
# import cascade
6-
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
7-
8-
from cascade.runtime.python_runtime import PythonClientSync, PythonRuntime
9-
from cascade.runtime.flink_runtime import FlinkClientSync, FlinkRuntime
10-
from deathstar_hotel_reservation.demo import DeathstarDemo, recommend, reserve, search_hotel, user_login
11-
import time
12-
import pytest
13-
14-
@pytest.mark.integration
15-
def test_deathstar_demo():
16-
ds = DeathstarDemo()
17-
ds.init_runtime(FlinkRuntime("deathstardemo-test", "dsd-out"))
18-
ds.runtime.run(run_async=True)
19-
print("Populating, press enter to go to the next step when done")
20-
ds.populate()
21-
22-
client = FlinkClientSync("deathstardemo-test", "dsd-out")
23-
input()
24-
print("testing user login")
25-
event = user_login()
26-
client.send(event)
27-
28-
input()
29-
print("testing reserve")
30-
event = reserve()
31-
client.send(event)
32-
33-
input()
34-
print("testing search")
35-
event = search_hotel()
36-
client.send(event)
37-
38-
input()
39-
print("testing recommend (distance)")
40-
time.sleep(0.5)
41-
event = recommend(req_param="distance")
42-
client.send(event)
43-
44-
input()
45-
print("testing recommend (price)")
46-
time.sleep(0.5)
47-
event = recommend(req_param="price")
48-
client.send(event)
49-
50-
print(client._futures)
51-
input()
52-
print("done!")
53-
print(client._futures)
54-
55-
def test_deathstar_demo_python():
56-
ds = DeathstarDemo()
57-
ds.init_runtime(PythonRuntime())
58-
ds.runtime.run()
59-
print("Populating, press enter to go to the next step when done")
60-
ds.populate()
61-
62-
time.sleep(0.1)
63-
64-
client = PythonClientSync(ds.runtime)
65-
print("testing user login")
66-
event = user_login()
67-
result = client.send(event)
68-
assert result == True
69-
event = user_login(succesfull=False)
70-
result = client.send(event)
71-
assert result == False
72-
73-
print("testing reserve")
74-
event = reserve()
75-
result = client.send(event)
76-
assert result == True
77-
78-
return
79-
print("testing search")
80-
event = search_hotel()
81-
result = client.send(event)
82-
print(result)
83-
84-
print("testing recommend (distance)")
85-
time.sleep(0.5)
86-
event = recommend(req_param="distance")
87-
result = client.send(event)
88-
print(result)
89-
90-
print("testing recommend (price)")
91-
time.sleep(0.5)
92-
event = recommend(req_param="price")
93-
result = client.send(event)
94-
print(result)
95-
96-
print("done!")
97-
98-
99-
if __name__ == "__main__":
100-
test_deathstar_demo()
2+
# import os
3+
# import sys
4+
5+
# # import cascade
6+
# sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../src")))
7+
8+
# from cascade.runtime.python_runtime import PythonClientSync, PythonRuntime
9+
# from cascade.runtime.flink_runtime import FlinkClientSync, FlinkRuntime
10+
# from deathstar_hotel_reservation.demo import DeathstarDemo, recommend, reserve, search_hotel, user_login
11+
# import time
12+
# import pytest
13+
14+
# @pytest.mark.integration
15+
# def test_deathstar_demo():
16+
# ds = DeathstarDemo()
17+
# ds.init_runtime(FlinkRuntime("deathstardemo-test", "dsd-out"))
18+
# ds.runtime.run(run_async=True)
19+
# print("Populating, press enter to go to the next step when done")
20+
# ds.populate()
21+
22+
# client = FlinkClientSync("deathstardemo-test", "dsd-out")
23+
# input()
24+
# print("testing user login")
25+
# event = user_login()
26+
# client.send(event)
27+
28+
# input()
29+
# print("testing reserve")
30+
# event = reserve()
31+
# client.send(event)
32+
33+
# input()
34+
# print("testing search")
35+
# event = search_hotel()
36+
# client.send(event)
37+
38+
# input()
39+
# print("testing recommend (distance)")
40+
# time.sleep(0.5)
41+
# event = recommend(req_param="distance")
42+
# client.send(event)
43+
44+
# input()
45+
# print("testing recommend (price)")
46+
# time.sleep(0.5)
47+
# event = recommend(req_param="price")
48+
# client.send(event)
49+
50+
# print(client._futures)
51+
# input()
52+
# print("done!")
53+
# print(client._futures)
54+
55+
# def test_deathstar_demo_python():
56+
# ds = DeathstarDemo()
57+
# ds.init_runtime(PythonRuntime())
58+
# ds.runtime.run()
59+
# print("Populating, press enter to go to the next step when done")
60+
# ds.populate()
61+
62+
# time.sleep(0.1)
63+
64+
# client = PythonClientSync(ds.runtime)
65+
# print("testing user login")
66+
# event = user_login()
67+
# result = client.send(event)
68+
# assert result == True
69+
# event = user_login(succesfull=False)
70+
# result = client.send(event)
71+
# assert result == False
72+
73+
# print("testing reserve")
74+
# event = reserve()
75+
# result = client.send(event)
76+
# assert result == True
77+
78+
# return
79+
# print("testing search")
80+
# event = search_hotel()
81+
# result = client.send(event)
82+
# print(result)
83+
84+
# print("testing recommend (distance)")
85+
# time.sleep(0.5)
86+
# event = recommend(req_param="distance")
87+
# result = client.send(event)
88+
# print(result)
89+
90+
# print("testing recommend (price)")
91+
# time.sleep(0.5)
92+
# event = recommend(req_param="price")
93+
# result = client.send(event)
94+
# print(result)
95+
96+
# print("done!")
97+
98+
99+
# if __name__ == "__main__":
100+
# test_deathstar_demo()

‎deathstar_movie_review/entities/frontend.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,11 @@ def compose_compiled_0(variable_map: dict[str, Any]):
3030

3131

3232
frontend_op = StatelessOperator(
33+
Frontend,
3334
{
3435
"empty": compose_compiled_0,
3536
},
36-
None
37+
{}
3738
)
3839

3940
def frontend_df_serial():
@@ -85,7 +86,7 @@ def frontend_df_serial():
8586
df.add_edge(Edge(n7a, n7))
8687
df.add_edge(Edge(n7, n8))
8788

88-
df.entry = n0
89+
df.entry = [n0]
8990
return df
9091

9192
def frontend_df_parallel():
@@ -135,5 +136,5 @@ def frontend_df_parallel():
135136
df.entry = [n1_a, n3_a, n5_a, n7]
136137
return df
137138

138-
frontend_op.dataflow = frontend_df_parallel()
139+
frontend_op.dataflows["compose"] = frontend_df_parallel()
139140

‎deathstar_movie_review/entities/text.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,16 @@ def upload_text_2_compiled_0(variable_map: dict[str, Any]):
1515
pass
1616

1717
text_op = StatelessOperator(
18+
Text,
1819
{
1920
"upload_text_2": upload_text_2_compiled_0
2021
},
21-
None
22+
{}
2223
)
2324

2425
df = DataFlow("upload_text")
2526
n0 = StatelessOpNode(text_op, InvokeMethod("upload_text_2"))
2627
n1 = OpNode(ComposeReview, InvokeMethod("upload_text"), read_key_from="review")
2728
df.add_edge(Edge(n0, n1))
28-
df.entry = n0
29-
text_op.dataflow = df
29+
df.entry = [n0]
30+
text_op.dataflows[df.name] = df

‎deathstar_movie_review/entities/unique_id.py‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
from typing import Any
22
import uuid
33
from cascade.dataflow.dataflow import DataFlow, InvokeMethod, OpNode, StatelessOpNode
4-
from cascade.dataflow.operator import StatelessOperator
4+
from cascade.dataflow.operator import Block, StatelessOperator
55
from deathstar_movie_review.entities.compose_review import ComposeReview
66

77

@@ -19,14 +19,15 @@ def upload_unique_compiled_0(variable_map: dict[str, Any]):
1919
variable_map["review_id"] = uuid.uuid1().int >> 64
2020

2121
unique_id_op = StatelessOperator(
22+
UniqueId,
2223
{
23-
"upload_unique": upload_unique_compiled_0,
24+
"upload_unique": Block(name="upload_unique", function_call=upload_unique_compiled_0, var_map_writes=["review_id"], var_map_reads=[]),
2425
},
25-
None
26+
{}
2627
)
2728

2829
df = DataFlow("upload_unique_id")
2930
n0 = StatelessOpNode(unique_id_op, InvokeMethod("upload_unique"))
3031
n1 = OpNode(ComposeReview, InvokeMethod("upload_unique_id"), read_key_from="review")
31-
df.entry = n0
32-
unique_id_op.dataflow = df
32+
df.entry = [n0]
33+
unique_id_op.dataflows[df.name] = df

‎deathstar_movie_review/entities/user.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
from typing import Any
22
from deathstar_movie_review.entities.compose_review import ComposeReview
33
from cascade.dataflow.dataflow import DataFlow, Edge, InvokeMethod, OpNode
4-
from cascade.dataflow.operator import StatefulOperator
4+
from cascade.dataflow.operator import Block, StatefulOperator
55

66

77
class User:

‎notebooks/dataflow_example.ipynb‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -313,12 +313,12 @@
313313
},
314314
{
315315
"cell_type": "code",
316-
"execution_count": 389,
316+
"execution_count": null,
317317
"metadata": {},
318318
"outputs": [],
319319
"source": [
320320
"from textwrap import indent\n",
321-
"from cascade.frontend.generator.generate_split_functions import GenerateSplittFunctions\n",
321+
"from cascade.frontend.generator.generate_split_functions import GenerateSplitFunctions\n",
322322
"from cascade.frontend.intermediate_representation import Block\n",
323323
"\n",
324324
"compiled_functions, df = GenerateSplittFunctions.generate_split_function_string(block_level_dataflow_graph)"
@@ -485,7 +485,7 @@
485485
},
486486
{
487487
"cell_type": "code",
488-
"execution_count": 456,
488+
"execution_count": null,
489489
"metadata": {},
490490
"outputs": [
491491
{
@@ -504,7 +504,7 @@
504504
}
505505
],
506506
"source": [
507-
"split_functions = GenerateSplittFunctions.generate(dataflow_graph)\n",
507+
"split_functions = GenerateSplitFunctions.generate(dataflow_graph)\n",
508508
"\n",
509509
"\n",
510510
"for i, split in enumerate(split_functions):\n",
@@ -617,7 +617,7 @@
617617
},
618618
{
619619
"cell_type": "code",
620-
"execution_count": 452,
620+
"execution_count": null,
621621
"metadata": {},
622622
"outputs": [
623623
{
@@ -639,7 +639,7 @@
639639
}
640640
],
641641
"source": [
642-
"split_functions = GenerateSplittFunctions.generate(dataflow_graph)\n",
642+
"split_functions = GenerateSplitFunctions.generate(dataflow_graph)\n",
643643
"\n",
644644
"\n",
645645
"for i, split in enumerate(split_functions):\n",

0 commit comments

Comments
 (0)