Good morning all and happy Monday slightly smiling face Does

Question

Good morning all and happy Monday slightly smiling face Does anyone here know how to stop Dagster from running ` op`s or ` asset` s with the MultiprocessExecutor and instead wait for one operation to finish before launching the next I m running a for loop over my asset keys and uploading files to them to GCS with the synchronous `gcloud sql export csv ` command But because of the synchronous nature of the command I end up uploading the first asset while all others fail silently disappointed ANY suggestions on how to make this implementation work in series synchronously would be much appreciated

Daniel Mosesson · Answer

You can use the in process executor

Daniel Mosesson · Answer

separately I think you might have an issue with your DAG if that is the issue you are running into

Harpal · Answer

Thats a good point < Daniel Mosesson> Do you mean method I haven t used an executor before but that s likely the best way to resolve this Are there any examples of your proposed solution I can have a look at on github or elsewhere

Daniel Mosesson · Answer

Yes exactly that For your job you do something like ` job executor def=in process executor ` <https docs dagster io apidocs execution dagster in process executor>

Harpal · Answer

Nice one I ll give that a try dagsir Thanks

Harpal · Answer

Source code and error logs as promised Code ```import subprocess from subprocess import Popen from dagster import AssetGroup asset from dagster dbt import dbt cli resource load assets from dbt project from moonfire dagster sector classification gcloud setup import setup gcloud DBT PROJECT DIR = dbt DBT PROFILE DIR = dbt config dbt assets = load assets from dbt project DBT PROJECT DIR select= +sector cls train set def csv assets for dbt assets dbt assets ret = for asset key in dbt assets 0 asset keys table name = asset key path 1 asset name=table name namespace= gcs non argument deps= asset key compute kind= gcs def asset setup gcloud need this to make sure auth works Same error ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete gcl export table comm = f gcloud sql export csv moonfire 01 <gs moonfire training data sector cls automatic table name csv> database=moonfire query= SELECT FROM public table name subprocess call gcl export table comm shell=True ret append asset return ret csv assets = csv assets for dbt assets dbt assets sector cls all dbt train test = AssetGroup dbt assets + csv assets resource defs= dbt dbt cli resource configured project dir DBT PROJECT DIR profiles dir DBT PROFILE DIR vars test split 0 2 build job sector cls ALL dbt train test ``` Error Logs ```Exporting Cloud SQL instance ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2910 gcs sector cls train set STEP OUTPUT Yielded output result of type Any Type check passed 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls train set Writing file at Users hdot vs code machine learning build dagster home storage a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls train set result 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2910 gcs sector cls train set ASSET MATERIALIZATION Materialized value gcs sector cls train set 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2910 gcs sector cls train set HANDLED OUTPUT Handled output result using IO manager io manager 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2910 gcs sector cls train set STEP SUCCESS Finished execution of step gcs sector cls train set in 3 63s ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2911 gcs sector cls test set STEP OUTPUT Yielded output result of type Any Type check passed ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls test set Writing file at Users hdot vs code machine learning build dagster home storage a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls test set result ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2911 gcs sector cls test set ASSET MATERIALIZATION Materialized value gcs sector cls test set 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2911 gcs sector cls test set HANDLED OUTPUT Handled output result using IO manager io manager 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2911 gcs sector cls test set STEP SUCCESS Finished execution of step gcs sector cls test set in 4 02s 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2905 gcs sector cls unk STEP OUTPUT Yielded output result of type Any Type check passed 2022 03 28 11 42 23 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls unk Writing file at Users hdot vs code machine learning build dagster home storage a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls unk result 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2912 gcs sector cls gcal STEP OUTPUT Yielded output result of type Any Type check passed 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls gcal Writing file at Users hdot vs code machine learning build dagster home storage a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls gcal result 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2903 gcs sector cls wak STEP OUTPUT Yielded output result of type Any Type check passed 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2913 gcs sector cls crunchbase data STEP OUTPUT Yielded output result of type Any Type check passed 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2905 gcs sector cls unk ASSET MATERIALIZATION Materialized value gcs sector cls unk 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls crunchbase data Writing file at Users hdot vs code machine learning build dagster home storage a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls crunchbase data result 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2907 gcs sector cls caf STEP OUTPUT Yielded output result of type Any Type check passed 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2905 gcs sector cls unk HANDLED OUTPUT Handled output result using IO manager io manager 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls caf Writing file at Users hdot vs code machine learning build dagster home storage a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls caf result 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls wak Writing file at Users hdot vs code machine learning build dagster home storage a44b0ba2 4d15 460a aa31 e00a87a51da1 gcs sector cls wak result 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2905 gcs sector cls unk STEP SUCCESS Finished execution of step gcs sector cls unk in 4 66s 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2912 gcs sector cls gcal ASSET MATERIALIZATION Materialized value gcs sector cls gcal 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2913 gcs sector cls crunchbase data ASSET MATERIALIZATION Materialized value gcs sector cls crunchbase data 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2912 gcs sector cls gcal HANDLED OUTPUT Handled output result using IO manager io manager 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2913 gcs sector cls crunchbase data HANDLED OUTPUT Handled output result using IO manager io manager 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2912 gcs sector cls gcal STEP SUCCESS Finished execution of step gcs sector cls gcal in 4 61s 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2907 gcs sector cls caf ASSET MATERIALIZATION Materialized value gcs sector cls caf 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2913 gcs sector cls crunchbase data STEP SUCCESS Finished execution of step gcs sector cls crunchbase data in 4 66s 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2907 gcs sector cls caf HANDLED OUTPUT Handled output result using IO manager io manager 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2907 gcs sector cls caf STEP SUCCESS Finished execution of step gcs sector cls caf in 4 69s 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2903 gcs sector cls wak ASSET MATERIALIZATION Materialized value gcs sector cls wak 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2903 gcs sector cls wak HANDLED OUTPUT Handled output result using IO manager io manager 2022 03 28 11 42 24 +0100 dagster DEBUG sector cls ALL dbt train test a44b0ba2 4d15 460a aa31 e00a87a51da1 2903 gcs sector cls wak STEP SUCCESS Finished execution of step gcs sector cls wak in 4 21s ```

Daniel Mosesson · Answer

Skimming a bit my hunch is that your `setup gcloud` and the following functions mean that you might benefit from having a ` resource` that handles that for you This would then be instantiated per `op` It also seems weird to me that all the assets are created in the same but to be honest I am just getting started with assets so I have no idea if that is the right way or if you would have better luck using an Op Factory or similar for something for this

Harpal · Answer

All very solid suggestions I m also new to software defined assets but I d love to hear what < owen> thinks is the right way If I can t get this gcloud one liner to work then I ll have to download all of the assets as csv files first and then upload them to GCS in a second step

Harpal · Answer

Unfortunately I haven t been able to run the aforementioned loop of asset s in series They keep launching subprocesses via the MultiprocessExecutor with their own pid and all see screenshot below for more details The input arguments for these assets are quite different from ops so I can t easily string them into a graph that I would with traditional op s Using ` build job sector cls ALL dbt train test executor def=ExecutorDefinition in process executor ` didn t change the output in any meaningful way < owen> Is it possible to run these assets in series waiting for one thread to finish before kicking off another or this way or is it not yet possible to restrict the number of workers threads in this way

owen · Answer

hi < Harpal> Daniel s suggestion is a reasonable one and the way to implement it with asset groups would be `from dagster import in process executor` and then ` build job executor def=in process executor ` The main drawback of this approach is that it means that if you ever add more assets into this group they ll also be bound by this parallelism constraint If that s not something you foresee doing then this is probably the best option

owen · Answer

the other option would be to take advantage of the fact that these assets definitions live on top of the underlying operational graph dagster currently has the concept of a `multi asset` although the APIs might shift a bit by dagster 1 0 which is essentially multiple assets mapped to a single op The body of that op could be a loop that iterates through each cli command one by one This would mean that you wouldn t have to mess around with the executor

Harpal · Answer

Hiya < owen> Thanks for suggesting the proper syntax I see what you mean It should be fine to have these assets bound to the parallelism constraint enforced ` build job sector cls ALL dbt train test executor def=in process executor ` The problem I m facing is that It s still spawning all of the processes with different `pid` numbers and all and running them at once This is visible from the screenshot below and the accompanying error message What could be going wrong to cause it to not wait for the first process to complete ```ASSET MATERIALIZATION Materialized value gcs sector cls crunchbase data 2022 03 28 18 53 08 +0100 dagster DEBUG sector cls ALL dbt train test a5580e24 8ee1 48ee a5a3 e63b8cab8d67 36874 gcs sector cls crunchbase data HANDLED OUTPUT Handled output result using IO manager io manager DEBUG POST sql v1beta4 projects moonfire instances moonfire 01 export alt=json HTTP 1 1 409 None DEBUG gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete Traceback most recent call last File Users hdot google cloud sdk lib googlecloudsdk calliope cli py line 987 in Execute resources = calliope command Run cli=self args=args File Users hdot google cloud sdk lib googlecloudsdk calliope backend py line 809 in Run resources = command instance Run args File Users hdot google cloud sdk lib surface sql export csv py line 64 in Run return export util RunCsvExportCommand args client File Users hdot google cloud sdk lib googlecloudsdk command lib sql export util py line 163 in RunCsvExportCommand return RunExportCommand args client csv export context File Users hdot google cloud sdk lib googlecloudsdk command lib sql export util py line 91 in RunExportCommand result operation = sql client instances Export export request File Users hdot google cloud sdk lib googlecloudsdk third party apis sqladmin v1beta4 sqladmin v1beta4 client py line 556 in Export return self RunMethod File Users hdot google cloud sdk lib third party apitools base py base api py line 737 in RunMethod return self ProcessHttpResponse method config http response request File Users hdot google cloud sdk lib third party apitools base py base api py line 743 in ProcessHttpResponse self ProcessHttpResponse method config http response request File Users hdot google cloud sdk lib third party apitools base py base api py line 609 in ProcessHttpResponse raise exceptions HttpError FromResponse apitools base py exceptions HttpConflictError HttpError accessing lt gt response lt vary Origin X Origin Referer content type application json charset=UTF 8 content encoding gzip date Mon 28 Mar 2022 17 53 08 GMT server ESF cache control private x xss protection 0 x frame options SAMEORIGIN x content type options nosniff alt svc h3= 443 ma=2592000 h3 29= 443 ma=2592000 h3 Q050= 443 ma=2592000 h3 Q046= 443 ma=2592000 h3 Q043= 443 ma=2592000 quic= 443 ma=2592000 v= 46 43 transfer encoding chunked status 409 gt content lt error code 409 message Operation failed because another operation was already in progress Try your request after the current operation is complete errors message Operation failed because another operation was already in progress Try your request after the current operation is complete domain global reason operationInProgress gt ERROR gcloud sql export csv HTTPError 409 Operation failed because another operation was already in progress Try your request after the current operation is complete ```

Harpal · Answer

I m really hoping that I can get these running in series as I d hate to have to pay for separate Extraction tool like fivetran just to do this one step in my DAG sweat smile

owen · Answer

hm it s really strange that the executor def argument isn t being respected here seems to indicate a bug on our side although looking at the code I don t see an immediate way for that to occur I ll dig into this more this afternoon sorry that you re running into this

owen · Answer

the multi asset solution would also work so I can throw together what that would look like as well

owen · Answer

``` def csv assets for dbt assets dbt assets outs = for asset key in dbt assets 0 asset keys table name = asset key path 1 outs table name = Out asset key=AssetKey gcs table name multi asset outs=outs non argument deps=dbt assets 0 asset keys compute kind= gcs def assets setup gcloud need this to make sure auth works for table name in outs keys gcl export table comm = f gcloud sql export csv moonfire 01 <gs moonfire training data sector cls automatic table name csv> database=moonfire query= SELECT FROM public table name subprocess call gcl export table comm shell=True yield Output table name table name return assets ```

owen · Answer

that would be something like this might be some random syntax errors in there but that s basically it

Harpal · Answer

No problemo Owen I ll keep beavering away from my end too Thanks so much for the pseudocode +1

owen · Answer

this solution just maps all your gcs csv file assets to a single operation which will do all of them in sequence

Harpal · Answer

I ve almost cracked it

Harpal · Answer

There seems to be an issue with the types accepted by `multi asset` ``` Error loading repository location repo py dagster check ParameterCheckError Param non argument deps is not one of frozenset set Got dict keys AssetKey sector cls haw AssetKey sector cls wak AssetKey sector cls unk AssetKey sector cls caf AssetKey sector cls train set AssetKey sector cls test set AssetKey sector cls gcal AssetKey sector cls crunchbase data which is type lt class dict keys gt ```

Harpal · Answer

We cool This may have fixed it ```non argument deps=set dbt assets 0 asset keys ```

owen · Answer

yep nice was just about to suggest that

Harpal · Answer

And it works AGAIN < owen> you are an absolute superhero

owen · Answer

< Harpal> just saw your message in < C01U5LFUZJS|dagster feedback> and right now we don t have a way to do that but one thing that would clean up your diagram a bit is ```def csv assets for dbt assets dbt assets outs = deps = for asset key in dbt assets 0 asset keys table name = asset key path 1 outs table name = Out asset key=AssetKey gcs table name deps table name = AssetKey table name multi asset outs=outs non argument deps=set dbt assets 0 asset keys compute kind= gcs internal asset deps=deps ``` basically just defining that each gcs asset actually only depends on its specific dbt model rather than every single one of them

Harpal · Answer

Hi < owen> I ve designed another job that need to upload assets to gcloud in a similar manner Is it possible to have two instance of the above functionality within the same `repo py` asset groups and the likes I seem to be having trouble getting a second job up and running and fear this may be why Is creating a second similar job in this manner possible If so how do I go about executing this

owen · Answer

hi < Harpal> what issue are you running into exactly if you just have mulitple invocations of csv assets for dbt assets what happens

Harpal · Answer

Hiya < owen> Apologies I kept it vague because I recall reading in the docs somewhere that it was not possible and was busy looking for workaround I found a pretty solid workaround for now which should tide me over for the time being crossed fingers I discovered that changing the input to the `select` argument to the function below allows one to pick precisely which dbt DAG to run for each job ```dbt assets = load assets from dbt project DBT PROJECT DIR select=f tag DATASET TYPE ``` Therefore I can run different DBT DAGs and save them to GCS by simply changing the value I provide to DATSET TYPE A global variable of input argument to some wrapper function should do the trick e g ```DATASET TYPE = os environ get DATASET TYPE DAG 1 ``` Is this poor practice in Dagster Because it seems to solve a lot of my problems without having to change any dagster code daggy 3d

owen · Answer

hi < Harpal> I wouldn t call this bad practice exactly but it does cut you off from being able to view the structure of both of these dbt dags in Dagit at the same time in general the biggest issues here would be UI concerns like that where we want to provide metadata about the structure of your job but the structure of your job is subject to change depending on the env var set

Harpal · Answer

Ahh I see what you mean Fortunately enough these two actually have a one node in common giving me one large DAG with two paths representing each tagged job So the DagitUI shows everything albeit a little cluttered sweat smile Clean as a whistle dagsir