Is there a way to force the sensor process to reload code or understand how long it will take before a code update happens? (for context) I would like to not rely on reties which seems like the current path without knowing

Hello, I think I have found a bug in

@asset

. Details in 🧵

Hey, can I run multiple dagster daemons on multiple nodes/servers?

Hi team, I am trying to test my project using

pytest

. I have

from project import my_repository

def test_repos():
    my_repository.load_all_definitions()

However, when I run this, I get the error

AttributeError: 'PendingRepositoryDefinition' object has no attribute 'load_all_definitions'

trying to update to dagster==1.1.6 from 0.15.9, running into an import error when the repository is being loaded:

DagsterImportError: Encountered ImportError: `cannot import name 'introspection_query' from 'graphql' (/databricks/python/lib/python3.8/site-packages/graphql/__init__.py)` while importing module repo. Local modules were resolved using the working directory `/Users/zachary.romer/Documents/empirico/etxlib`. If another working directory should be used, please explicitly specify the appropriate path using the `-d` or `--working-directory` for CLI based targets or the `working_directory` configuration option for workspace targets.

it looks like the

gql

library that's referenced in

dagster-graphql

isn't pinned, so it didn't get updated when I updated my environment. does anyone know what minimum version of

gql

is required by

dagster-graphql

? looks like a breaking change in that library is causing issues for me, but I'm having some dependency resolution issues with other requirements when I try pinning it to 3.0.0+

Hey team, how do I fix these errors?

2022-12-15 00:14:26 +0000 - dagster.daemon.QueuedRunCoordinatorDaemon - ERROR - Another QUEUED_RUN_COORDINATOR daemon is still sending heartbeats. You likely │
have multiple daemon processes running at once, which is not supported. Last heartbeat daemon id: 37ed50b2-39e7-437b-8823-bc4e36e6ea11, Current daemon_id: 5bf│
7e12e-1371-47c8-94ad-33b285b4d0a3                                                                                                                             │
2022-12-15 00:14:26 +0000 - dagster.daemon.SchedulerDaemon - ERROR - Another SCHEDULER daemon is still sending heartbeats. You likely have multiple daemon pro│
cesses running at once, which is not supported. Last heartbeat daemon id: 37ed50b2-39e7-437b-8823-bc4e36e6ea11, Current daemon_id: 5bf7e12e-1371-47c8-94ad-33b│
285b4d0a3                                                                                                                                                     │
2022-12-15 00:14:26 +0000 - dagster.daemon.BackfillDaemon - ERROR - Another BACKFILL daemon is still sending heartbeats. You likely have multiple daemon proce│
sses running at once, which is not supported. Last heartbeat daemon id: 37ed50b2-39e7-437b-8823-bc4e36e6ea11, Current daemon_id: 5bf7e12e-1371-47c8-94ad-33b28│
5b4d0a3                                                                                                                                                       │
2022-12-15 00:14:31 +0000 - dagster.daemon.SensorDaemon - ERROR - Another SENSOR daemon is still sending heartbeats. You likely have multiple daemon processes│
 running at once, which is not supported. Last heartbeat daemon id: 37ed50b2-39e7-437b-8823-bc4e36e6ea11, Current daemon_id: 5bf7e12e-1371-47c8-94ad-33b285b4d│
0a3

Hi, is there a way to immediately return and skip the rest of the Ops? With success job status?

Dynamic Queuing of a graph based on run parameters received from frontend: Is it possible to queue run of a graph based on the parameters received in the run_config.

Hi, I want to generate an error message in dagster job runs and it should show failure status in the list of run jobs. Thank

Hi, I have the following run_status_sensor to catch all job that have been cancelled

@run_status_sensor(run_status=DagsterRunStatus.CANCELED)
def job_canceled_sensor(context: RunStatusSensorContext):
    prometheus = PrometheusClient()
    status = 'canceled'
    logger = get_dagster_logger(name='JOB_STATUS_SENSOR')
    <http://logger.info|logger.info>(f"job_canceled_sensor - {context.dagster_run.job_name} - {status}")
    prometheus.change_job_state(context.dagster_run.job_name, status)
    prometheus.pushadd_to_gateway(job_name=context.dagster_run.job_name)

But then when I terminate the job via the UI, the job state is CANCELED but the sensor is not triggered. Am I doing something wrong?

Hello Team I'm getting an intermittent issue of run not starting. Runs are starting in separate containers. Sometimes this issue comes more frequently.

Hello, I just updated Dagster to 1.1.6 version but I am now unable to materialize any asset manually using the

Materialize all / Materialize selected

button. Details in 🧵

Hey, I'm trying out a POC of dagster currently and comparing to something like Luigi and Airflow where it is really simple to chain tasks dynamically given the input. Is it possible to do something similar in dagster? I want to do something like:

def run_node(node):
    if node.has_child_nodes():
        for child_node in node.child_nodes():
            yield run_node(child_node)

    yield {"res": my_func(node)}


def my_job():
    graph = load_graph()
    for node in graph:
        run_node(node)

Hey guys! My team's pretty new to Dagster. We're trying to load assets from an external source (Hightouch, we're using reverse ETL) into Dagster. I wrote a function that calls Hightouch api and maps all data syncs to

assets

. Said function is called once in a

repository

along with

dbt

and

airbyte

build functions. All looks ok in Dagit (upstreams resoved, etc). However, when I click materialize before the

op

that builds said

asset

the parent function is called several times again. Am I doing something wrong? Or is this an expected behavior? I've looked at airbyte assets and I can see a

CacheableAssetsDefinition

instead of

AssetsDefinition

used there. Should I just re-implement this for Hightouch? I couldn't find an example that would show how to load assets from external source, only how to explicitly define them in Dagster. But for us at the moment, this would cause a lot of boilerplate, so we're testing other aproaches.

Does someone have examples of using the UPathIOManager? I checked the docs and it's not that obvious to me atm

I have another random question, as I am using resources and io_managers, is there a way to specify with perhaps typing or something specific resource types? I am publishing a library with many resources and io managers and I am worried users will confuse or mix up the required resource key with the wrong resources, since the only thing that connects them seems to me is just a key string

Hi everyone, new to Dagster so still learning a lot. Has anyone here worked with Delta Lake and Spark Structured Streaming in Dagster? Essentially I have pipelines that process S3 file resources using Spark structured streaming, then transform them in bronze, silver, gold, Delta tables. The streams process more on the second-minute interval in microbatches, and some only run a few times a day and shut off. Before I start writing my own IO managers, does anyone have any resources they could point me to for Structured Streaming and Delta? Thanks!

currently working on an ETL for a multi tenant application. Each tenant has an underlying accounting system that we need to map into our data model. So Tenant 1 and Tenant 2 could share accounting system 1 and Tenant 3 could be accounting system 2. Within each accounting system the ETL will be the same but the destinations of the assests will be different. Right now I have a couple functions that I pass in a tenant name and create all the assets for the given tenant so the psuedo code is something like

def create_bronze_assets(tenant: str): 
    @asset(...)
    def some_asset()
    return [some_asset]

# in __init__.py that has the new defs = Definitions line
tenant_one = create_accounting_system_assets('tenant_one')
tenant_two = create_accouting_system_assets('tenant_two')
tenant_three = create_other_accounting_system_assets('tenant_three')
defs = Defintions(...)

Depending on the tenant we would materialize each set of assets once a day. I'm looking at making the assets partitioned by day for that. Then I got looking into multi dimensional partitions and possible thinking about using the tenant name as the first key of the partition and the date as the second. Any thoughts on this idea or better ways to handle this multi tenancy?

Hello guys, How should we set the

jsonCredentialsEnvvar

for

gcsComputeLogManager

? I got the credentials json, converted to string using json.dumps() in python, and passed this string in the helm chart. However it returns:

ValueError: ('Could not deserialize key data. The data may be in an incorrect format, it may be encrypted with an unsupported algorithm, or it may be an unsupported key type).

How should I pass this json ?

Is the

asset_reconciliation_sensor

supposed to work with cross-repository asset dependencies? I get a

KeyError

from the sensor when it detects a change in an upstream asset in a different repository, see stacktrace:

KeyError: AssetKey(['my', 'cool', 'asset'])
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/errors.py", line 199, in user_code_error_boundary
    yield
  File "/app/venv/lib/python3.9/site-packages/dagster/_grpc/impl.py", line 324, in get_external_sensor_execution
    return sensor_def.evaluate_tick(sensor_context)
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/sensor_definition.py", line 425, in evaluate_tick
    result = list(self._evaluation_fn(context))
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/sensor_definition.py", line 589, in _wrapped_fn
    result = fn(context)
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/asset_reconciliation_sensor.py", line 858, in sensor_fn
    run_requests, updated_cursor = reconcile(
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/asset_reconciliation_sensor.py", line 694, in reconcile
    ) = determine_asset_partitions_to_reconcile(
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/asset_reconciliation_sensor.py", line 362, in determine_asset_partitions_to_reconcile
    all(
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/asset_reconciliation_sensor.py", line 374, in <genexpr>
    instance_queryer.is_reconciled(
  File "/app/venv/lib/python3.9/site-packages/dagster/_utils/cached_method.py", line 59, in helper
    result = method(self, *args, **kwargs)
  File "/app/venv/lib/python3.9/site-packages/dagster/_utils/caching_instance_queryer.py", line 205, in is_reconciled
    for parent in asset_graph.get_parents_partitions(
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/asset_graph.py", line 257, in get_parents_partitions
    for parent_asset_key in self.get_parents(asset_key):
  File "/app/venv/lib/python3.9/site-packages/dagster/_core/definitions/asset_graph.py", line 172, in get_parents
    return self.asset_dep_graph["upstream"][asset_key]

And the `RepositoryDefinition`; keeping in mind that all assets fetched from dbt depend on upstream `SourceAsset`s.

@repository
def my_repo():
    project = "my_project"
    return [
        slack_on_run_failure,
        *with_resources(**generate_dbt_repo(project, DEPLOYMENT_NAME)),
        build_asset_reconciliation_sensor(
            AssetSelection.all(),
            name="reconciliation_sensor",
        )
    ]

Hello, I am trying to figure out how to register jobs in Dagster based on information accessible via a service call. For example, for FooJob, our customers use a UI to add the jobs which get saved to our database, and I can access them via an API call to a service. I'm having trouble figuring out how to register jobs without hardcoding them. I'm aware of this "assets" abstraction but that doesnt look right for my use case. Is this the best way to create a dynamic number of jobs?

@repository
def foo_jobs():
	all_jobs = []
	for customer_id in rpc_client.get_all_customers():
		for job_name in rpc_client.get_all_foo_jobs(customer_id):
			all_jobs.append(
				dagster.JobDefinition( ??? )
			)
	return all_jobs

Or do I need to write some kind of generate to write out the python code with the annotations? I am aware of this "assets" concept as well, but I don't think that will work because these jobs have different schedules, and I need direct control of the graph for each job (different FooJobs could have different graphs based on information known only at runtime).

hey all, what is the proper way to setup nodeAffinity or nodeSelector for

dagster-step

jobs? Is there a way to configure it in the Helm Chart or it should be passed in the k8s_executor?

How can I access metadata in an op? AttributeError: 'OpExecutionContext' object has no attribute 'metadata'

Hi, how long does it take for new environment variables set in Dagster cloud UI to take effect? I have a job run that complains about missing env variables, I added that in UI and rerun the job, it still complains about missing env variables. I’m sure I selected the right scope. Is there anything else I need to do to make it work.

For an asset with a large amount of depdendencies, .e.g.

D depends on C, C on A + B

When I want to convert the asset to a job in order to define a schedule for it, shall I put all A+B+C+D to

define_asset_job

?

Hello guys, How should the

jsonCredentialsEnvvar

for

gcsComputeLogManager

be ? Should it be the path to the json file? Should it be the json file dumped as a string? If it is the path to the file, should it be set inside each different workspace(inside each different docker image containing my repo's) ? Could I stock my credential.json in a GCP bucket and use its address as

jsonCredentialsEnvvar

? Thanks !

How can i configurate assets to use different tokens (specify run_config into repository)? btw in ore repository use same assets?

Hi all, I have a question regarding user code deployments on k8s. Does anyone know why the image used for dagster runs would not update despite updates to the to the user code deployment subchart? I'm using the helm chart with dagster version 1.1.5 in all pods. Dagit is configured with:

workspace:
    enabled: true
    servers:
      - host: "k8s-my-pipeline"
        port: 3030
        name: "my-pipeline"

When I update the image tag in the values.yaml of the subchart and redeploy it, the code server image updates correctly but the jobs launched from dagit still use the original image.

deployments:
  - name: "k8s-my-pipeline"
    image:
      repository: "my_pipeline_image"
      tag: changed_tag
      pullPolicy: Always
    dagsterApiGrpcArgs:
      - "--python-file"
      - "/opt/my_pipeline/repository.py"
    port: 3030
    includeConfigInLaunchedRuns:
      enabled: True

Hey everyone, I'm trying to get logs from a module that uses the python logging module, I tried everything on the docs https://docs.dagster.io/concepts/logging/python-logging#configuring-a-python-log-level but nothing works, where/how am I supposed to get these logs? running dagit locally I don't see it in the dagit process logs, or in the Run UI at all, even with all the log types checked

Trying to think of a design pattern for this scenario: • assets produced using a file system IO Manager -> writing to local disk for speed • one all assets have materialized, some of these are then uploaded to a database -> but this would be triggered as a separate job I'm thinking • Should each asset then have a second asset with a different IOManager (DB upload)? • Then have a "define_asset_job" which just selects those assets that will be uploaded?