Create dynamic assets on runtime

[AlejandroUPC]

It feels like Dagster has been pushing for software-defined assed directions (which is a right direction in my opinion) but I am still struggling to wrap my head about how to do some stuff being used to a more op/job mindset.

Right now it's kind of easy to imlpement the following logic using ops/jobs/graphs:

1. List all of the credentials in a path.
2. For each credential, create an op (and the downstream) to do some computations that reads data from a database.
3. At the end materialize the data in a file.

The pseudo code is something along the lines (please understand that is pseudo code)

import os
from dagster import op, job, DynamicOut, DynamicOutput


@op(out=DynamicOut())
def get_client_credentials(context):
    """
    Creates downstream op for every client whose credentials are on the creds/ path.
    """
    for creds_file in os.listdir("/creds/"):
        with open(creds_file, 'r') as fr:
            yield DynamicOut(value=fr.read(),mapping_key=creds_file)

@op
def get_invoices(creds:str) -> pd.DataFrame:
    """
    Extracts credentials to connect to a database and returns a dataframe as the result from the invoice's query.
    """
    user,pwd = creds.split(",")
    connection = connect(user,pwd)
    return connection.query(f"SELECT * FROM invoices WHERE client_name={creds_file}")

@op
def get_max_invoice(context,invoices:pd.DataFrame) -> None:
    """
    Stores in S3 on a specific path per client the max invoice in the invoices dataframe.
    """
    bucket = boto.s3()
    invoices_max = invoices.max()
    bucket.write(path=f"{context.get_mapping_key()}/max",data=invoices_max)


@job
def put_max_to_s3():
     clients_creds = get_client_credentials()
     client_creds.map(get_invoices).map(get_max_invoice)

This has several benefits, so I can add as many clients as I want to the folder creds/ without having to modify any of my code, the downstream jobs run in parallel somewhat, I can easil identify which ops for each client are failing and some of the dynamic created ops will run until the end even if a specific client fails.

Now I am not sure how to implement this althought I have been looking at some discussions such as #9559 but I have really struggled to understand the examples and definitions given.

And also saw some comments such as

• software-defined assets currently don't support dynamic asset keys, and it's unlikely we will in the future. However, we will likely add support for dynamic partition keys, which is the solution proposed in the issue description above.

What's the right mindset to approach this based on the software asset definitions?

EDIT

Just came across #11045, but still would be interested on having a more similiar example to the one I posted, given that if you really need to generate multiple linked assets it seems a bit complex as show in 11045, and still have some flaws: not sure how to access the context inside the asset (given that it should be passed to the factory?) What happens if I want to create dynamically two assets that are dependent between them? I cant pass them as argument to the function (Python does not allow the creation of dynamic arguments in its functions).

Similarly #15087 shows what seem to be a too complex way to do it given how easy is to do so with ops.

[sryza]

@AlejandroUPC I believe this is the issue where we're tracking this functionality: #9559.

Does what's described there match what you're looking for?

[AlejandroUPC]

That's indeed a good shot together with this article, but I am not sure if it really fits the case. I am still not convinced by any of the answers in there or I see that there's a final consensus reached, if you really feel its a duplicated then feel free to close it, thanks.

Like let's say I have a file folder with different confi_files per client that need to be processed according to the config (imagine creds to connect to sql and a query to run), I want the ability to do exactly the same as my main example or your example with the files, is that the correct approach? I am not sure, given that if I the partitions are static they will be overwritten every time its run? (e.g if my config_files dir has client a,b,c and it runs 1h for a day the same partition will be overwritten although the content of the data they are pulling might be ifferent).

[AlejandroUPC]

To be honest it feels quite confusing and I am not sure if I fully understand the benefits vs risks of using an asset factory vs StaticPartitions for my case @sryza feel free to close it and thanks

[sryza]

In general, if you have large numbers of objects with the same structure, we recommend using partitions instead of using a separate asset for each object. This makes it easier to perform bulk operations like backfills, as well as makes the UI more scalable to navigate.
Here's some content on this: #12061.

[AlejandroUPC]

I am satisfied with that, a little bit sceptical but I will investigate how to create jobs from that and schedule them. Thank you a lot for your time and patience

[mihirsamdarshi]

Hi @sryza I have a certain use-case, and I'm not sure how the suggested answer might be able to help me.

One of my assets is a table that contains metadata regarding a number of other datasets and the list of tables that those assets will end up in. That list of assets may be updated from time to time. I'd like to dynamically create a number of assets from those datasets, using an op or something similar. The assets all have different schemas. Is that possible to do with Dagster?

Edit: Never mind I found the discussion at #9559