Hey all Does anyone know how to save the assets loaded from

Question

Hey all Does anyone know how to save the assets loaded from DBT to a csv file I m using software defined assets and the `AssetGroup ` method to gather all of my tables as described in the and can now see the assets in the dagit UI I would like to use these assets in an op to write them as a csv file and then perform this as a job Better yet I can write a sensor to listen to the asset key that is generated by this job But how do I actually pull the postgresql table into dagster from the metadata surrounding the asset Are there any relevant docs available because I could not find anything relevant under or anywhere on the dagster website Thanks in advance See comments for a code snippet

Harpal · Answer

```dbt assets = load assets from dbt project DBT PROJECT DIR select= +sector cls train set sector cls train test = AssetGroup dbt assets resource defs= dbt dbt cli resource configured project dir DBT PROJECT DIR profiles dir DBT PROFILE DIR vars test split 0 2 ```

jamie · Answer

< owen>

owen · Answer

hi < Harpal> Let me know if I m misunderstanding your question but is the basic end result that you re trying to achieve 1 run dbt to update a bunch of tables 2 copy the contents of those updated tables into csv files 3 do some sort of further processing on those csv files I think the key abstraction that you need here is the IOManager The IOManager defines how dagster will store in memory outputs from assets ops as well as how dagster will load the those stored values as inputs to downstream assets ops For things produced by dbt by default nothing additional will need to be stored by dagster for each output because the data is already stored in the database As for loading there are a bunch of different ways to load the contents of a database table as an in memory object We have a very lightweight example in the example This implements the protocol of to store dbt output do nothing to load dbt as input fire query against the configured database and load as a dataframe For your usecase depending on your needs you might implement that same protocol or something like to store dbt output write table to csv file to load dbt as input load that csv file as a dataframe

owen · Answer

there s an io manager key parameter to the load assets from dbt project function which you can map to an io manager that you define in your resource defs on your asset group

owen · Answer

I also don t see an immediate need to break this out into a group of assets + a separate job my first instinct would be to create python assets instead of ops that depend on the dbt assets there s an example of that in the repo i linked above

Harpal · Answer

Hiya < owen> Thanks again for your deeply insightful responses Allow me to clarify my usecase with a bit of useful context I have lots of tables in a postgres database hosted by google cloud sql I have a DBT DAG that transforms that data Each of the tables created in the multi stage tranformation all written in SQL DBT are shown as assets thanks to the code snippet above I want to be able to take to take these materialised tables that simultaneously exist in my postgres db and displayed in the DagitUI take the tables by their asset key and store these tables s csv files in google cloud storage GCS Then use those csv files in GCS in downstream training jobs I ve done some snooping around and have discovered that gcloud CLI has a handy one liner solution to move a postgres table from cloud sql to GCS as a csv file `gcloud sql export csv moonfire 01 database=moonfire query= SELECT FROM public sector cls test set ` I can t say I know much about IOManagers But i ve managed to carve together a proposed workaround I SHOULD be able to listen for the assets outputted from the dagster dbt job with a dagster sensor and then trigger a job that runs the handy gcloud one liner This will essentially move the relevant tables from postgres to GCS What I don t know is Whether this proposed workaround practical or would I be missing out on some awesome dagster optimisations How feasible this solution is given the experimental status of some of the methods I hope that helps clarify things Looking forward to your thoughts and suggestions dagsir

owen · Answer

ah I see the solution you re describing would work but I think it would be more ergonomic in the long run to write everything as assets and keep things in a single `AssetGroup` The way I would conceptualize this is that each csv file on GCS is in fact its own separate asset distinct from the table in your database You could manually specify these csv assets or you could automatically generate one for each asset created in your dbt project I threw together some code that should basically do that but haven t tested it so buyer beware stuck out tongue ```dbt assets = load assets from dbt project 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 run one liner for this table name ret append asset return ret csv assets = csv assets for dbt assets dbt assets asset group = AssetGroup dbt assets + csv assets ``` Your downstream training jobs depending on how they re implemented might also make sense to be written as a sequence of assets that depends on these csv assets but it s possible that it makes more sense to keep those in job style syntax

owen · Answer

Anyway the main benefit of this sort of solution over your proposed solution is that the csv files will be linked both in the metadata that Dagster tracks and in the UI to the dbt models that they depend on which makes it easier to reason about what s going on rather than depending on an implicit link defined by an asset sensor The csv files will also be treated as a separate consumable entity for potential downstream assets and it will be much easier to notice issues of the form dbt ran successfully but the csv conversion failed

Harpal · Answer

< owen> OK I m convinced Your solution is definitely the way forward

Harpal · Answer

I ve adapted your pseudo code and I think Dagster in unhappy with the `ret` variable returning a list of ` asset` objects ```2022 03 23 21 18 18 +0000 dagster daemon SensorDaemon WARNING Could not load location repo py to check for sensors due to the following error dagster check CheckError Member of list mismatches type Expected lt class dagster core asset defs asset AssetsDefinition gt Got lt dagster core asset defs asset AssetsDefinition object at 0x2a0014880 gt lt dagster core asset defs asset AssetsDefinition object at 0x2a0014eb0 gt lt dagster core asset defs asset AssetsDefinition object at 0x16f0e2220 gt lt dagster core asset defs asset AssetsDefinition object at 0x16f0e23d0 gt lt dagster core asset defs asset AssetsDefinition object at 0x16f0e2580 gt lt dagster core asset defs asset AssetsDefinition object at 0x16f0e2730 gt lt dagster core asset defs asset AssetsDefinition object at 0x16f0e28e0 gt lt dagster core asset defs asset AssetsDefinition object at 0x16f0e2a90 gt lt dagster core asset defs asset AssetsDefinition object at 0x16f0e2c40 gt of type lt class list gt ```

Harpal · Answer

The stack trace seems to agree Throwing the following error ```Stack Trace File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster grpc server py line 212 in init self loaded repositories = LoadedRepositories File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster grpc server py line 97 in init loadable targets = get loadable targets File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster grpc utils py line 27 in get loadable targets else loadable targets from python file python file working directory File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster core workspace autodiscovery py line 18 in loadable targets from python file loaded module = load python file python file working directory File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster core code pointer py line 79 in load python file return import module from path module name python file File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster seven init py line 47 in import module from path spec loader exec module module File lt frozen importlib bootstrap external gt line 850 in exec module File lt frozen importlib bootstrap gt line 228 in call with frames removed File moonfire dagster repo py line 13 in lt module gt from moonfire dagster sector classification import sector cls all dbt assets File Users hdot vs code machine learning moonfire dagster sector classification sector cls all dbt assets py line 34 in lt module gt asset group = AssetGroup File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster core asset defs asset group py line 90 in new check list param assets assets of type=AssetsDefinition File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster check init py line 368 in list param return check list items obj list of type File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster check init py line 449 in check list items raise CheckError ```

Harpal · Answer

Ooops Almost forgot the adapted code ```import subprocess 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 setup gcloud 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 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 ret append asset return ret csv assets = csv assets for dbt assets dbt assets asset group = 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 ```

owen · Answer

ah yeah it should just be `dbt assets + csv assets` not ` dbt assets + csv assets `

Harpal · Answer

Ahhh my bad I was following the example in <https docs dagster io concepts assets software defined assets building jobs that materialize assets|these docs>

owen · Answer

ah yeah there s some trickiness where some functions return lists of AssetsDefinitions and regular python assets are just singular AssetsDefinition objects Leads to some inconsistent looking code confusion

Harpal · Answer

Gotcha I m almost there Just need to figure out what s causing this `DagsterInvalidDefinitionError` Sorry but the error message doesn t give me much to go by

Harpal · Answer

```2022 03 23 21 30 34 +0000 dagster daemon SensorDaemon WARNING Could not load location repo py to check for sensors due to the following error dagster core errors DagsterInvalidDefinitionError Bad return value from repository construction function all elements of list must be of type JobDefinition GraphDefinition PipelineDefinition PartitionSetDefinition ScheduleDefinition or SensorDefinition Got value of type lt class module gt at index 25 Stack Trace File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster grpc server py line 212 in init self loaded repositories = LoadedRepositories File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster grpc server py line 97 in init loadable targets = get loadable targets File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster grpc utils py line 27 in get loadable targets else loadable targets from python file python file working directory File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster core workspace autodiscovery py line 18 in loadable targets from python file loaded module = load python file python file working directory File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster core code pointer py line 79 in load python file return import module from path module name python file File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster seven init py line 47 in import module from path spec loader exec module module File lt frozen importlib bootstrap external gt line 850 in exec module File lt frozen importlib bootstrap gt line 228 in call with frames removed File moonfire dagster repo py line 32 in lt module gt def repo File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster core definitions decorators repository py line 242 in repository return Repository name File Users hdot pyenv versions 3 9 8 envs machine learning 3 9 8 lib python3 9 site packages dagster core definitions decorators repository py line 65 in call raise DagsterInvalidDefinitionError ```

Harpal · Answer

It seems to be upset with repo py ``` File moonfire dagster repo py line 32 in lt module gt def repo ```

owen · Answer

do you mind sharing what your repo function looks like I agree that the error message isn t all that illuminating

owen · Answer

seems like you might have an `import some local filename` at the top of repo py and then inside repo py you have `some local filename`as a member of that list

Harpal · Answer

Oooooooooooooh it works