Hey guys, Databricks question here: is it possible/ does it make sense to execute an op within a Databricks context on a cluster and transform it somehow into an asset and make it visible in dagit? Basically right now, we’re only triggering runs for already deployed Databricks jobs with Dagster that store data on S3, without all the asset benefits. Do you have any experiences with that setup?

It looks like your image didn’t install the “celery” package. Is it possible to share your Dockerfile?

Hey Alex - when you run dagit you need to tell it where to load your code from. There are some example arguments to dagit here to do this: https://docs.dagster.io/concepts/code-locations#deploying-and-loading-code-locations

Do assets need to be able to fit into memory? I'm guessing practically they do and that partitioning is the recommended path to avoid assets which are too large for in memory processing?

Hi, is there an equivalent in Dagster to Argo's work avoidance? https://argoproj.github.io/argo-workflows/work-avoidance/ I looked into docs and the experimentla feature of versioning seems like a similiar feature with extra steps: https://docs.dagster.io/guides/dagster/memoization Is there any other option to skip already materialized upstream assets based on some conditions? I have DBT models in mind.

Hi Team, question about using Definition in Dagster. Previously we are able to have multiple repository in Dagster to group different jobs. But when I try to use Definitions, it says it cannot have more than one Definitions object defined at module scope, how could I still use Definition to group jobs like what I did in the past?

Hi all, i succeeded to deploy dagster with helm chart (with postgresql enabled), but when i try to use an external database my daemon and dagit pod still in init state, any default values examples with external database ?

Hi! I am trying to define an asset using a graph and I can't get the asset key to work as I'd hoped. The pattern is this:

data1 = SourceAsset(key='data1') 
data1 = SourceAsset(key='data2') 

@op(out=Out(io_manager_key='database_io_manager',asset_key=AssetKey('my_table_name),)
)
def infutor_address_tci(tci: str):
    return tci


@op(
    ins={
        'start': In(Nothing),
        'tch': In()
        },
    out=Out(
        io_manager_key='kinetica_io_manager',
        asset_key=AssetKey(['kinetica', 'address', 'infutor_address']),
        metadata={
            'load_options': {
                'batch_size': '50000',
                # tch only has a subset of data, hence the links to out of bounds (901....)
                'columns_to_load': '29, 8, 10..20,24,901..902,,32,904',
                'file_type': 'csv',
                'text_delimiter': '\t',
                'text_has_header': 'false',
                'text_quote_character': '~',
                'error_handling': 'permissive',
            }
        }
    )
)
def infutor_address_tch(tch:str):
    return tch


@op(required_resource_keys={'db'})
def update_infutor_address(context: OpExecutionContext, table):
    # remove lat/lon not at geocode rooftop, line interpolated, or parcel_centroid level.
    context.resources.db.execute_sql(f"""
        UPDATE {table.name}
        SET
            latitude=NULL,
            longitude=NULL
        where
            (latitude is not null or
            longitude is not null) and 
            nvl(geocode not in ('01','02'),true)
        """
                                     )
    return None


@graph()
def infutor_address(tci: str, tch: str):
    table = infutor_address_tch(tch, start=infutor_address_tci(tci))
    return update_infutor_address(table)


infutor_address = AssetsDefinition.from_graph(
    infutor_address,
    keys_by_input_name={
        'tci': AssetKey(['s3', 'infutor', 'tci']),
        'tch': AssetKey(['s3', 'infutor', 'tch'])
    },
    key_prefix=['kinetica', 'address'])

hi guys, I am creating a MVP from an existing pipeline to validate dagster as possible alternative airflow, step functions, I loved the tool, super easy to start but I am struggle to create a production environment. Is there an example where I have multiple user repositories?

Hi, Dagster family! I tried searching for a simple answer in the docs and on slack but couldn't find what I wanted- hoping one of you kind souls can point me the right way... I want to run some ops in sequence (they are dependent on each other), they need to pass a dictionary of values to each other. Additional twist: the ops need to come from an op factory! Example of what I'm looking to do:

Hey everyone, I am working on a flow where an asset - A creates partitions on hourly basis, and an asset B which will depend on a set of partitions of asset A, i want to configure a range of partitions of asset A while running asset B on dagit UI, is this possible?

Hey, is there a way to rename the repository that is created by a

Definition

object? Like with using the

@repository

decorator? In the source code I can see it gets a default value of

name=SINGLETON_REPOSITORY_NAME

Can we override it somehow or is this only possible using the decorator?

Hey everyone, I have a small question regarding logging. For Software Defined Assets, we should always use the name of the function itself in the

asset_key

parameter of

AssetMaterialization

and

AssetObservation

is it?

Is there already a solution for dealing with errors like this ?

dagster._check.CheckError: Invariant failed. Description: Assets defined for node 'DailyJob' have a partitions_def of Daily, starting 2022-10-20 UTC., but job 'HourlyJob' has non-matching partitions_def of Hourly, starting 2022-10-20-00:00 UTC..

Hi Team, I’m trying to get a solution to monitor sensors that fail (for example when they run the checks to decide which jobs to run). I haven’t found things like

hooks

or

run_failure_sensor

there are for jobs/ops. What is the best Approach for this?

I want to add a Daemon to Dagit without editing the workspace.yml directly. Could there be a GraphQL endpoint for this? How about a UI for adding locations similar to Code Locations in Cloud?

Hi Jim - what I suspect is happening here is a version issue where the celery workers are using an earlier version of Dagster that doesn't recognize that step

It looks like I can't pass a simple string to op's arguments, but then how do I make the next op dependent on the first one and only run if the first one succeeds.

I'm running into an issue where: •

dagit -w workspace.yaml

works •

dagit -w ../other/folder/workspace.yaml

works but `dagit -w worspace.yaml -w ../other/folder/workspace.yaml`hangs after starting the code server for the first package Doesn't matter which order I put them in, both workspaces have only one python package specified (with attendant executable paths). Same issue if I try putting the workspaces together. No additional logs show up when I add

--log-level trace

Any idea what could be going on?

Hi, my Dagster Instance is consuming a lot of RAM just to stay alive (about 2.9GB), is there any way to avoid that? Maybe it's related to the repositories, I have 3 of them.

Hi, can I use an Op to persist and Asset? I need to iterate a dataframe and hit an API foreach item.

@asset
def my_dataframe() -> Output[pandas.DataFrame]:
    df = pandas.read_csv('some_path.csv')
    return df

@op
def persist_df(my_dataframe):
    for index, row in my_dataframe.iterrows():
        # request here
        pass


@job
def schedulable_job():
    persist_df()

Hey everyone. I'm hoping someone can point me towards some examples and best practices for using Dagster with other languages (R in particular). I have several ETL pipelines written entirely in R that could really benefit from using a data orchestrator and Dagster is my first choice. I'm trying to find a balance between rewriting my entire process in Python versus just calling a bunch of R scripts sequentially. Any insight here is appreciated!

Team, hello. We have our own triggering and retry logic in our data pipeline, but we only want to use the visualisation part of the Dagstar. to understand what stage the pipelines have been at so far. can we do this using dagster?

Hello everyone, When using the class

UPathIOManager

the method

context.add_output_metadata()

does not work, rather we have to implement

UPathIOManager.get_metadata()

. It would be helpful if the docs explicitly stated that calling

context.add_output_metadata()

from

UPathIOManager.dump_to_path()

does not work.

Hey everyone, I'm looking to migrate an existing ETL flow to dagster but I'm not sure how to configure/implement dagster in my specific situation using best practices. At the moment, i have two meta tables. One meta table contains the dependency logic of the different tables (assets in dagster I assume) in the form of a column 'Source' and 'Target'. When one 'source' have multiple targets, the table contains multiple rows for that source. Further, the second table links each table with a stored procedure which do the necessary transformation steps. Now I want to implement this logic into dagster. Which concepts should I use for the different components of the ETL logic and how can I generate, based on the meta tables, the different assets/ops/jobs in dagster? Thank you in advance!

Hi, I have suddenly started to get the following error on my Dagster deployment running in Kubernetes. The Daemon restarts and I see:

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) could not translate host name "$HOST_NAME" to address: Temporary failure in name resolution

This started happening all of a sudden and my DB is running so that shouldn't be an issue. I am running Dagster version 1.1.3 and HELM chart 1.1.3. Is this a known bug in that version and updating should fix it or is it something else?

Hi everyone, I’m new to dagster and I just joined to ask this question, since I’ve spent a while searching through the documentation and couldn’t find a clear answer. I have an hourly partitioned asset like so:

@asset(partitions_def=HourlyPartitionsDefinition(start_date="2022-01-01-00:00"))
def blog_posts(context) -> List[Dict]:
    partition_datetime_str = context.asset_partition_key_for_output()
    hour = datetime.fromisoformat(partition_datetime_str)
    posts = fetch_blog_posts_from_external_api(hour_when_posted=hour)
    return posts

I can use the above asset as input to other assets - that’s all nice and fine. But what I would also like to be able to do is to run a job each time a new partition of

blog_posts

has been materialized in order to publish the data to yet another external API. I assumed I would be able to do this fairly easily using an asset sensor, like so:

@asset_sensor(asset_key=AssetKey("blog_posts"), job=my_job)
def new_materialized_blog_posts_sensor(context, asset_event):
    yield RunRequest(
        run_key=context.cursor,
        run_config={
            "ops": {
                "my_op": {
                    "config": {
                        "asset_key_path": asset_event.dagster_event.asset_key.path,
                        "partition_key": asset_event.dagster_event.partition,
                    }
                }
            }
        },
    )

This triggers

my_job

and that job executes

my_op

which receives the asset_key and partition. So it seems like I’m just a step away. But the problem I have is: how do I access the underlying data? Do I need to work directly with the

IOManager

class to load the input? If so, how do initialise the correct IO manager based on what has been configured for the asset storage? Once I have the IO manager, I should be able to access the data using the approach for testing:

context = build_input_context(asset_key=asset_key, partition_key=partition_key)
blog_posts = manager.load_input(context)

But this seems a bit hacky (the recommendation is to use this for tests only, I think). So what is the best way I can access the materialized asset partition in my other job?

Hi! I have a question during using of `asset_sensor()`: Is it possible to get metadata values I stored with

AssetMaterialization

event? For instance, I have `AssetMaterialization`:

AssetMaterialization(
    asset_key=AssetKey("my_key"),
    description="Update asset",
    metadata={
        "date": "2022-12-01"
    },
)

Then I would like to get these value in

asset_sensor

method:

@asset_sensor(asset_key=AssetKey("my_key"), job=dbx_to_druid_general_job)
def internal_bi_bid_funnel_etl_job_sensor(context: SensorEvaluationContext, asset_event: EventLogEntry):
    ...

hi @Oleksandr (Alex) Dimov! For the run_config question, the solution of creating those ops is reasonable. Another option to make that configuration globally available without having to create those ops would be using make_values_resource For your second question of how to get things to execute after meltano_run_op, you can use nothing dependencies. In your case, that would look like changing your op decorator to

@op(ins={"cc_id": In(), "tbl": In(), "start_after": In(Nothing))
def cf_run_event_processing(context, cc_id, tbl):

then your job would have

meltano_done = meltano_run_op(...)()
cc_id, cache_key = cf_run_event_processing(..., start_after=meltano_done)

Hi Team, I’m trying to retry failed jobs on a later time and trying to avoid them taking a place in the queue while the wait. RequestRetry and Retry policy keep the job in the queue. 2 questions: • What is the best way to retry jobs from a run_failure_sensor? • What’s the best generic way to rerun a job at a later time? Thanks in advance for the help