19913d85-f662-4518-a9a1-7adc1793dfd4.png){kind=small}
Hi Guys! I am very new with Dagster, actually I a...
# announcementsi
Istvan Darvas
11/19/2020, 2:15 PMHi Guys! I am very new with Dagster, actually I am evaluating with this cool technology. I am trying to setup a pipeline with external parameters. So i have read the documentation and watch some presentation on youtube and I saw some solution for external (not scheduled) based pipeline invocation. Well I am trying to simulate this scenario, but it is not clear for me how can I give parameters to this event based pipeline during development, because I always get error 😉. I am going to put some code, screenshot into the thread. Any help would be appreciated.
Istvan Darvas
11/19/2020, 2:16 PM Copy code
@pipeline(
mode_defs=[
ModeDefinition("Live", resource_defs={"df_reader": dataframe_reader_for_pgsql}),
],
input_defs=[InputDefinition(name="candidate", dagster_type=dict)],
)
def ignition_aggregate_pipeline(candidate):
candidate = { # This would be my preset
"device_inventory_id": 1,
"device_id_type": 2,
"device_id_name": "58909",
"proc_until": "2020-11-18 15:38:23+01",
"receive_time": "2020-11-19 12:35:34+01",
}
ignition_agg_base = new_ignitions_in_range(candidate)Istvan Darvas
11/19/2020, 2:17 PMI try to invoke this:
Istvan Darvas
11/19/2020, 2:17 PM Copy code
@solid(
required_resource_keys={"df_reader"},
input_defs=[InputDefinition(name="candidate", dagster_type=dict)],
output_defs=[OutputDefinition(name="debug", dagster_type=int)],
)
def new_ignitions_in_range(context, candidate: dict) -> int:Istvan Darvas
11/19/2020, 2:18 PMI always get this error:
Istvan Darvas
11/19/2020, 2:18 PMdagster.core.errors.DagsterInvalidDefinitionError: In @pipeline ignition_aggregate_pipeline, received invalid type <class 'dict'> for input "candidate" (at position 0) in solid invocation "new_ignitions_in_range". Must pass the output from previous solid invocations or inputs to the composition function as inputs when invoking solids during composition.
Istvan Darvas
11/19/2020, 2:20 PMMy question is how can I develop with some predefined paramters? How can I pass it when I deployed my pipeline (deployed: I am not here yet 🙂 )
s
sandy
11/19/2020, 2:51 PMhi @Istvan Darvas - how are you executing your pipeline? with
execute_pipelinei
Istvan Darvas
11/19/2020, 2:53 PMHi @sandy!
Istvan Darvas
11/19/2020, 2:53 PM Copy code
if __name__ == "__main__":
run_config = {}
with open("dagster-test2_resources.yaml", 'r') as yaml_cfg:
try:
run_config = yaml.safe_load(yaml_cfg)
except yaml.YAMLError as exc:
print(exc)
execute_pipeline(ignition_aggregate_pipeline, run_config=run_config)Istvan Darvas
11/19/2020, 2:53 PMI put it into the code, because i would like to use pycharm during development...so i can debug
Istvan Darvas
11/19/2020, 2:54 PMthis is why i would like to also setup a preset for development
Istvan Darvas
11/19/2020, 2:55 PMbut I am still not sure, i can pass parameters as arguments to the decorated pipeline function
s
sandy
11/19/2020, 4:19 PMone way you could accomplish this is:
Copy code
@pipeline(
mode_defs=[ModeDefinition("Live", resource_defs={"df_reader": dataframe_reader_for_pgsql}),],
)
def ignition_aggregate_pipeline():
ignition_agg_base = new_ignitions_in_range()
@solid(
required_resource_keys={"df_reader"},
output_defs=[OutputDefinition(name="debug", dagster_type=int)],
config_schema={
"device_inventory_id": int,
"device_id_type": int,
"device_id_name": str,
"proc_until": str,
"receive_time": str,
},
)
def new_ignitions_in_range(context) -> int:
candidate = context.solid_config
execute_pipeline(
ignition_aggregate_pipeline,
run_config={
"solids": {
"new_ignitions_in_range": {
"device_inventory_id": 1,
"device_id_type": 2,
"device_id_name": "58909",
"proc_until": "2020-11-18 15:38:23+01",
"receive_time": "2020-11-19 12:35:34+01",
}
}
},
)i
Istvan Darvas
11/19/2020, 5:06 PMmaybe "this is the way" 😄
Istvan Darvas
11/19/2020, 5:06 PMIstvan Darvas
11/19/2020, 5:06 PMI created a test solid just returning my initial, development preset
Istvan Darvas
11/19/2020, 5:07 PMif there is another, better way and there is someone who are not afraid to share it, it would be appreciated
c
cat
11/19/2020, 5:28 PM@Istvan Darvas was wondering if the code snippet sandy sent above works for you?
i
Istvan Darvas
11/19/2020, 5:32 PM@cat @sandy I have to admit, i did not noticed 😞 I am going to test it immediately and thanks you warn me again.
Istvan Darvas
11/19/2020, 5:39 PMThis will work, I can see the logic thanks, I just need to rewrite that function, so in this case I need get these parameters from the context pass around
Istvan Darvas
11/19/2020, 5:39 PMi am going to do this way
Istvan Darvas
11/19/2020, 5:39 PMThanks guys
s
sandy
11/19/2020, 5:40 PMglad to help!
👍 1
i
Istvan Darvas
11/19/2020, 5:51 PMIt's really working 😉
Istvan Darvas
11/19/2020, 5:51 PMIstvan Darvas
11/19/2020, 5:51 PM@sandy thanks dude!
c
cat
11/19/2020, 8:58 PMawesome 👍 🎉
2 Views