Hi All, I was interested to try out the freshness capability, but noticed that this wasn’t appying to DBT-generated assets, adding the following into the AssetsDefinition function call in _dbt_nodes_to_assets_ seemed to get this working properly, is this on anyone’s radar currently or is it a case of making a git pr, making a general use case rather than hardcoding may be more tricky?

freshness_policies_by_key={
    asset_key: FreshnessPolicy(maximum_lag_minutes=45) for asset_key, (input_name, _) in (asset_ins | asset_outs).items()
}

I am trying to push lineage and metadata to Apache Atlas whenever a materialization happens. Any ideas for how to do this? I've looked at OpHooks and AssetSensors, but not sure what the best practice would be, I'd want it to happen on all asset materializations in a repostiory

Hey guys, Really enjoying the added observability as I start to migrate existing jobs to the asset paradigm! I may have missed it, but was curious what is the best way to add default tags to the assets loaded from DBT? I would like to ensure that any DBT asset, on materialization, adds the default

dagster-k8s/config

tag to its underlying op. It seems like tags are only defined on the jog level and not necessarily on the asset level?

Hello friends, If I add metadata to my materialization using

context.add_input_metadata

, how do I access the metadata in another run. I currently use this to get metadata that was added using

context.add_output_metadata

latest_materialization_event = self.init_context.instance.get_latest_materialization_events(
            [asset_key]
        ).get(asset_key)

        <http://context.log.info|context.log.info>(latest_materialization_event)
        if latest_materialization_event:
            materialization = (
                latest_materialization_event.dagster_event.event_specific_data.materialization
            )
            metadata = {entry.label: entry.entry_data for entry in materialization.metadata_entries}

Wondering if there is a way to get

context.add_input_metadata

metadata?

Hello experts, I have a question about the best practice when working with SQL databases. Let's imagine I want to insert some data returned by some public API into SQLite. I can think of two ways of doing this: 1. Define an op to run the sql insert or update queries 2. Define an io manager to do the same. Is there a preferred way?

Missing something obvious here I’m sure: I have a graph definition from which I produce a job with

to_job()

. The graph has one op in it which is async and must be awaited therefore the graph function must be async as well obviously but this arrangement throws

dagster._core.errors.DagsterInvalidDefinitionError: @graph 'nba_trends_graph' returned problematic value of type <class 'coroutine'>. Expected return value from invoked solid or dict mapping output name to return values from invoked solids

If I am running `dagster-daemon`/`dagit` on one machine, how can I

dagster job launch -j some_job --config-json {...}

Do I need to pass the gRPC host/port? Where do I find those, are they shown somewhere on the dagit ui?

Hi all 🙂 having a go with multidimension partitions, I'm a little lost on how I can achieve something. I have two assets

partition_def = StaticPartitionDefinition(['0','1','2'])
@asset(
partition_def=partition_def
)
def a(): return 'a'

@asset(
partition_def=partition_def
)
def b(): return 'b'

now I want to introduce a third asset c that produces a partition for the cross product of the other two. so

cross_part = MultiPartitionsDefinition({
    'a': partition_def,
    'b': partition_def
})
@asset(
partition_def=partition_def
)
def c(a, b): return a+b

I guess I need to implement two `PartitionMapping`s? one which takes the first part of the multikey and the other that takes the second part and then supply like

partition_mapping={'a': APartitionMapping(), 'b': BPartitionMapping()}

- is that the correct approach?

We are getting the grpcio issue while installing. I am trying to install dagster on Mac M1 using different versions of python 3.11, 3.10 and 3.8 and the installation fails at grpcio. pip is also upgraded. Kindly help.

I have a service that provides a python API that allows a client to open a WebSocket "subscription" to a non-continuous stream of data (i.e., events happen at somewhat unpredictable times--kind of like an RSS feed, though not identical). Part of the API allows to specify a function that is triggered whenever an event occurs on this subscription. But in order not to miss any events, I have to continuously monitor the subscription. So, for now I have a python program with a

while (True)

loop listening to this subscription and sending the data provided by the event to a particular function. The plan is that part of that function will write a file to disc, which will then trigger a Dagster sensor. I would prefer to be able to have Dagster listen to this subscription and fire an op whenever an event happens on that subscription. Is there something like an IOManager example that could be helpful for developing something like this? Thanks in advance.

Hello, I'm testing the

step_isolated_job

mode of execution in Kubernetes, and I'm running into an issue: When a job is launched, the run will create a pod for each op (or asset). This pod will run the command

dagster api execute_run

. My issue is that I use poetry, and to run

dagster

, the command to run

dagster

is

poetry run dagster ....

. My pod thus crashes and the job stops. For the user code deployment pod, it's easy because I can specify my own command to run the Dagster GRPC api, but since pods of the run are launched and managed by Dagster, I have no control on them. Is there a way to specify some kind of prefix so that Dagster launches the correct command inside the pod? Thanks in advance! 🙂

When executing some conected Ops with the

in_process_executor

, are the outputs/inputs serialized/deserialized between steps? Or since they are in the same process, they are pushed from one Op to the other in memory?

Hello there, hoping this is the right place to ask about it. I’m new to Dagster, so I might not be aware of lot of works going on but so far I really like what has been done here. I’m looking at the architecture of Dagster and the scalability of it. I’m surprised about the

Dagster deamon

which can’t (per the doc) handle replicas for the moment. As it’s running sensors / schedule etc… I was wondering: 1. If there was benchmarks of workload it can handle in //? 2. Any plan to make it scalable? (Didn’t found any issues mentioning it)

Is there a way to test

run_monitoring

without having to wait for the situation where a run worker crashes to arise on it’s own?

Hi, does dagster store logs from sensor somewhere?

Using dagster 1.0.16 I have an asset that looks like

@asset(group_name=ASSET_GROUP)
def file_uuid() -> str:
    file_uuid = utils.file.create_file_uuid()
    return file_uuid

from dagit I can materialize this on its own when clicking the "Materialize" button. If I shift+click the "Materialize" button, the launchpad opens but errors for missing config are thrown. If I then click "Scaffold Missing Config", config for unrelated ops/assets is populated in the launchpad. Why am I forced to scaffold config for assets/ops that are not part of the lineage of the asset I want to materialize? I also noticed in the bottom right of the launchpad under the header "Assets (Ops)" I see all my repo's assets instead of just the one asset I selected when attempting to materialize. Thoughts?

Hello Dagster Expert, Is there way to grouping (or create sub-group under workspace?) For example, we want to group jobs on this image under different theme

I am getting this error when using the built-in asset reconciliation sensor, what's happening?

Hi there, I have been experimenting with the Observable sources, and was interested how best to automatically schedule the running of ‘observe sources’. It seems to be well set up to do the materialization of assets using a scheduled job like this:

ScheduleDefinition(cron_schedule="* * * * *",
                                        job = define_asset_job("update_sources", selection=AssetSelection.all()))

but I cant seem to find an equivalent for observing sources on a schedule?

Hi, We have failure hooks setup on job level to notify us of op failures within the job to slack. Within our function to send the slack notification, we parse values such as context.op_name, context.job_name, context.op_exception etc (from context: HookContext) to be sent as additional values in our slack alert. In one of the jobs, we have retries set for op failures. We notice that when the op failures reaches max retries specified, the hook is triggered and we get the alert for failure. But we observe that during this scenario , the context.op_exception field is null and we dont get it in our slack alert message. Is this behavior expected?

How to get the schedule name in a success sensor ?

Hi dagster support, we believe we've found an issue with running backfills for AssetJobs in 1.1.1 - when there is resource config that is required, the

materialize all

button always shows the launchpad, rather than the UI for launching backfills 🧵

Hey all 🙂 Getting some grpc errors in dagit, anything to worry about?

Operation name: AssetGraphLiveQuery

Message: maximum recursion depth exceeded while calling a Python object

Path: ["assetNodes",26,"projectedLogicalVersion"]

Locations: [{"line":44,"column":3}]

Hi Is there a way to set a config to be used for production and for development (slack notifications)

(Shorter rewrite) Does an approach like this fit? Dealing with a database source that has revisions to historical data. The downstream asset's a parquet dataset, partitioned by day. It's a simple base table replication for staging. The whole thing works like a selective backfill. Once a day: • query the source database to get a "manifest" for the table - grouped by partition key (date), row count, max modify date for each. Usually there will be the new day's data plus some small subset of changed previous days. • For each partition row in the manifest, check if it's missing or changed in the partitioned asset (parquet folders) • Run the asset's op for each partition to materialize it (query to extract that day, store and overwrite that parquet file) Seems like it should be possible, I'm just trying to figure out which kinds of objects to use to schedule this initial single query activity that then spawns ops for arbitrary partition keys in the asset as needed. It gets confusing when mixing non-partitioned and partitioned jobs, not sure I'm looking at it the right way. It seems like it might work better to not try to integrate with Dagster's asset+job partitioning at this level, and just have the asset op do everything I described internally.

Is there a way to get dagster to output logs from the regular logging lib without configuring it per logger? I have a lot of code I would like to run but the usuall python logging lib gets ignored. i.e - logger = logging.getLogger(_ name_)

Hi all! Has anyone already come across how to handle an error if RUN is not created due to a configuration error when starting schedullers? I use run_failure_sensor but it doesn't catch such errors. Tell me, how can I catch?

Is it possible to change the name of the IO manager? 😄 '

Also is it possible to call an Op from an asset?