How can I create a single asset sensor with resources The do dagster #ask-community

How can I create a single asset sensor with resour...

geoHeil

04/14/2022, 1:45 PM

How can I create a single asset sensor with resources? The documentation showcases: https://docs.dagster.io/concepts/partitions-schedules-sensors/sensors#asset-sensors 1. when following along with it I get an exception that the run_configuration does not match in particular the filed

read_materialization

is not accepted:

Copy code

Sensor daemon caught an error for sensor my_asset_sensor : dagster.core.errors.DagsterInvalidConfigError: Error in config for job
    Error 1: Received unexpected config entry "read_materialization" at path root:ops. Expected: "{ baz_scd2_asset?: { config?: { assets?: { input_partitions?: { } output_partitions?: { } } } inputs?: { baz_asset?: Any } outputs?: [{ result?: { json: { path: String } pickle: { path: String } } }] } }".

How can this be fixed? Further: 2. For me it is quite confusing, that resources need to be specified multiple times: - for the asset grup - for any job instanciated from an asset group And perhaps it is a misconception, but so far I understand that for each asset that needs to automatically update based on an upstream changed asset I need an accompanying sensor. Therefore following along the builder pattern I need a job https://github.com/geoHeil/dagster-ssh-demo/blob/master/SSH_DEMO/sensors/sftp_sensor_asset_real.py#L270 Can I somehow combine more steps in a single job that trigger updates directly (like ops) for assets? When trying to not specify any (additional) run_config from the asset_sensor the job fails as well as the pyspark resource (which is required to compute the asset) is not available. this leads me to the following questions: - how can I get this to work (resources in sensors in a nice way)? Will this change/get simpler in the future? Can I somehow combine more steps in a single job that trigger updates directly (like ops) for assets? - is it correct (having so many jobs i.e. one for each asset that should auto-update)

geoHeil

04/14/2022, 2:10 PM

And (3) (perhaps this is a bug in dagster or a misunderstanding on my side): executing this asset manually from the asset page (without the definition of the job) works totally fine. But trying to manually run the asset from the jobs launch butten fails due to the missing resource. This is quite confusing for me.

sandy

04/14/2022, 5:28 PM

if I understand correctly, you're only using the pyspark resource inside the asset / IO manager code, not inside the sensor itself. is that right? if so, the pyspark resource should go on the AssetGroup, and you shouldn't need to reference it at all inside the sensor

sandy

04/14/2022, 5:29 PM

regarding

read_materialization

, we don't currently offer the ability for a sensor to supply configuration when creating a

RunRequest

for an assets job, but @sean is working on it

geoHeil

04/14/2022, 9:17 PM

correct - and it is already registered in the asset group. I do not want to send additional configuration for the run request. But still it is not found. So how can I trigger an update for an asset (from a sensor) which needs the pyspark resource?

sandy

04/14/2022, 9:21 PM

when you say it's not found, are you able to share the error you're hitting and the code that's triggering the error?

geoHeil

04/14/2022, 9:22 PM

The error is: nknown resource

pyspark

. Specify

pyspark

as a required resource on the compute / config function that accessed it. and the code https://github.com/geoHeil/dagster-ssh-demo/blob/master/SSH_DEMO/sensors/sftp_sensor_asset_real.py#L277 but if you want I could briefly also join on zoom.

sandy

04/14/2022, 9:25 PM

do you possibly have the full stack trace available?

geoHeil

04/14/2022, 9:26 PM

Copy code

``
`dagster.core.errors.DagsterUnknownResourceError: Unknown resource `pyspark`. Specify `pyspark` as a required resource on the compute / config function that accessed it.`
  `File "/usr/local/Caskroom/miniconda/base/envs/dagster-ssh-demo/lib/python3.9/site-packages/dagster/core/execution/plan/execute_plan.py", line 232, in dagster_event_sequence_for_step`
    `for step_event in check.generator(step_events):`
  `File "/usr/local/Caskroom/miniconda/base/envs/dagster-ssh-demo/lib/python3.9/site-packages/dagster/core/execution/plan/execute_step.py", line 306, in core_dagster_event_sequence_for_step`
    `for event_or_input_value in ensure_gen(step_input.source.load_input_object(step_context)):`
  `File "/usr/local/Caskroom/miniconda/base/envs/dagster-ssh-demo/lib/python3.9/site-packages/dagster/core/execution/plan/inputs.py", line 163, in load_input_object`
    `yield _load_input_with_input_manager(loader, load_input_context)`
  `File "/usr/local/Caskroom/miniconda/base/envs/dagster-ssh-demo/lib/python3.9/site-packages/dagster/core/execution/plan/inputs.py", line 587, in _load_input_with_input_manager`
    `value = input_manager.load_input(context)`
  `File "/usr/local/Caskroom/miniconda/base/envs/dagster-ssh-demo/lib/python3.9/site-packages/dagster/core/storage/root_input_manager.py", line 162, in load_input`
    `return self._load_fn(context)`
  `File "/usr/local/Caskroom/miniconda/base/envs/dagster-ssh-demo/lib/python3.9/site-packages/dagster/core/asset_defs/assets_job.py", line 280, in _root_manager`
    `return io_manager.load_input(input_context_with_upstream)`
  `File "/Users/geoheil/development/tma/dagster/SSH_DEMO/SSH_DEMO/resources/duckdb_parquet_io_manager.py", line 72, in load_input`
    `return context.resources.pyspark.spark_session.read.parquet(path)`
  `File "/usr/local/Caskroom/miniconda/base/envs/dagster-ssh-demo/lib/python3.9/site-packages/dagster/core/definitions/resource_definition.py", line 410, in __getattr__`
    `raise DagsterUnknownResourceError(attr)`


``

sandy

04/14/2022, 9:29 PM

ahh, I see what's going on here. I believe this is a bug. I will get a fix out for this in our next release

geoHeil

04/14/2022, 9:31 PM

So I would not expect to call build_resources given the fact that the assetGroup`s resource_defs already contain pyspark.

sandy

04/14/2022, 9:32 PM

That's right - you shouldn't need to call

build_resources

for

pyspark

geoHeil

04/14/2022, 9:33 PM

cool then I am looking forward to a fix. Please can you share the issue with me once you have created it.

geoHeil

04/14/2022, 10:05 PM

One more question: (2) written above: Is there a way to have assets auto update without writing a (separate) sensor/job for each of these?

geoHeil

04/14/2022, 10:07 PM

(assuming only a basic asset changed event without any fancy extras i.e. the one already connected from the lineage)

sandy

04/14/2022, 10:33 PM

we don't currently have this, but we're interested in adding it in the future

sandy

04/14/2022, 10:57 PM

here's a PR with a fix for the resources issue: https://github.com/dagster-io/dagster/pull/7459

🎉 1

geoHeil

04/15/2022, 3:48 AM

Awesome! Regarding the PR: I observed you already pushed some changes. Do they already fix the issue? (even if for now perhaps without additional tests) I would then think about using/testing with the new PR-version of dagster today.

sandy

04/15/2022, 4:13 AM

The changes I pushed should fix the issue

🎉 1

✅ 1

21 Views

Open in Slack

Previous Next