Hi Team, I’m trying to test out environment variable locally in dagit. I put in a .env file and in the same folder run dagit, but still see env not set error.

$ cat .env
# .env
SNOWFLAKE_USERNAME=user@example.com
$ dagit
2022-12-13 09:48:21 -0800 - dagit - INFO - Serving dagit on <http://127.0.0.1:3000> in process 58295

Is there any indication that dagit has recognized the file and ingested it correctly? The error I see is

dagster._core.errors.DagsterInvalidConfigError: Error in config for resource snowflake
Error 1: Post processing at path root:config:user of original value {'env': 'SNOWFLAKE_USERNAME'} failed:
dagster._config.errors.PostProcessingError: You have attempted to fetch the environment variable "SNOWFLAKE_USERNAME" which is not set. In order for this execution to succeed it must be set in this environment.

I have an

@op

yielding `DynamicOutput`s that's taking 30m to yield 5600 outputs, taking about 0.5s between each "Yielded output" log message both when it started and when it was finishing. Is this expected? Running on k8s. I wouldn't mind except it doesn't seem like the outputs start running until the op finishes. The op is pretty simple:

@dataclass
class Test:
   test_id: str
   # etc

@op(out=DynamicOut(Test))
def yield_each_test(test_list: List[Test]):
    for test in test_list:
        yield DynamicOutput(value=test, mapping_key=test.test_id)

Is there a way to structure the

@op

more efficiently?

Hi Folks, I'm trying to launch a hybrid agent on ECS but keep getting this error

dagster._check.CheckError: Failure condition: Couldn't import module dagster_cloud.instance when attempting to load the class dagster_cloud.instance.DagsterCloudAgentInstance

Hello all. Is it possible to add new jobs to a repository without having to restart the container? I want to make a client that can dynamically assemble a job visually, then submit it. I can't seem to figure out if this is even possible.

Should I switch my sensor off when running backfill or Dagster will handle the race condition?

hello, is it possible to monitor multiple DagsterRunStatus with this decorator

run_status_sensor

? is multiple functions have to be defined with each DagsterRunStatus?

Hi team, I have a question on dagit init containers (dagster 0.15.8). It seems to create an init container for each user repo. If one fails, it prevents dagit from starting up properly.

until nslookup metrics-repo; do echo waiting for user service; sleep 2; done

Does it mean that if one user repo is problematic, dagit won’ boot up?

What’s the minimum delay I could expect between triggering a Job run and the first Op starting execution? And what’s the minimum delay I could expect between a Op finishing execution and the next Op in the Job starting execution? Context: Every time a new user of my application is created, I want to trigger a Job run that will scrape this user’s data from multiple sources and populate our database. I need to populate my database within 30 seconds of user creation. This is achievable with my current implementation of orchestrating this pipeline via a python function. I want to make sure that migrating my orchestration logic to Dagster will not add significant time to this process. Ideally I’d keep the additional time under 5 seconds. Is this achievable with Dagster?

Hi all, Is there a possibility to set default sensor, schedule run status for some env? I found that The default status can be overridden from Dagit or via the GraphQL API, bu is there option to change the default status to running for all the sensors and schedules. What I wan to achieve is having on some env different status than in the other without always writing:

@sensor(job=asset_job, default_status=SOME_ENV_CONSTANCE)
def my_sensor():
    ...

Can i combine two jobs in a schedule? I have some assets and jobs that should run on those assets. I am not ready to use sensors just yet for the jobs that i want to run on the assets. it's a bit more ad-hoc than a simple sensor. But the job/graph depends on a specific asset (moving the value of an asset to a external system)

Is it possible to

freeze

the

create_timestamp

|

update_timestamp

when creating a run for testing purposes? I want to have it fixed but its always using the current timestamp

with instance_for_test() as instance:
    for run in MOCK_RECORDS:
        cast(DagsterInstance, instance).add_run(
            PipelineRun(
                pipeline_name=run.pipeline_run.pipeline_name,
                run_id=run.pipeline_run.run_id,
                status=run.pipeline_run.status,
                )
            )

[RunRecord(storage_id=3, pipeline_run=DagsterRun(pipeline_name='job_dependency', run_id='coo', run_config={}, mode=None, asset_selection=None, solid_selection=None, solids_to_execute=None, step_keys_to_execute=None, status=<DagsterRunStatus.SUCCESS: 'SUCCESS'>, tags={}, root_run_id=None, parent_run_id=None, pipeline_snapshot_id=None, execution_plan_snapshot_id=None, external_pipeline_origin=None, pipeline_code_origin=None), create_timestamp=datetime.datetime(2022, 12, 14, 9, 14, 20), update_timestamp=datetime.datetime(2022, 12, 14, 9, 14, 20), start_time=None, end_time=None), ... ]

I want to filter by said dates or is there another way?

Is there a way to specify a

typing.Union

return type from an

@op

that doesn’t cause Dagster to raise a

DagsterInvalidDefinitionError

? I have an

op

that should return one of two custom types depending on some of the config passed, but specifying it in the return type signature or

Out(Union[type1, type2])

raises this error. The workaround I found was to define another output type (below) but it feels unnecessary for this type of use case.

type1_or_type2 = DagsterType(
    name="type1_or_type2",
    type_check_fn=lambda _, x: isinstance(x, type1) or isinstance(x, type2),
)

Hi All. I'm using a couple of custom PartitionMappings to map between some different Multipartitions. It works great! Since 1.1.6, I've gotten the warning

DeprecationWarning: Non-built-in PartitionMappings, such as MarketToDatePartitionMap are deprecated and will not work with asset reconciliation. The built-in...

Am I still going to able to have custom PartitionMappings in the future? Or is this feature just going away for asset reconciliation sensors?

Hi everyone, is it possible to run Dagster in debug mode and use breakpoints? I am using VSCode

Hi, can someone suggest which io-manager is best for passing file-objects from one op to another. Currently I am using the default io-manger fs_io_manger and its not letting me pass the file-obj. I tried mem_io_manager but it then ask to use fs_io_manger

Is it possible to define an asset without an input? I know you can do so with Ops by creating one in the form of:

@Op(ins={"start": In(Nothing)})

but

start

is not a key that is available in

ins

of an

asset

. I have a pipeline that is like:

ingest_dataset1 -> ingest_dataset2 -> start_processing_data

in the form of:

asset -> asset -> op.

The assets do not depend on each other, I just want one to run after the other. I could also use:

op -> op -> op

but it is to my understanding that an Op should be used when performing some type of transformation or logic whereas an asset is signifies something which is stored to a persistent storage.

I'm trying to set up a config mapping for ops. Some ops I am reusing and creating an alias

@op
def generate_sample1() -> pd.DataFrame:
    # <http://context.log.info|context.log.info>("config_param: " + context.op_config["config_param"])
    return sample_data()

@graph
def graph_multi_sample():
    n_samples = 5
    samples = []
    for i in range(n_samples):
        samples.append(generate_sample1.alias(f"sample_{i}")())
    return concat_samples(samples)

job_from_graph = graph_multi_sample.to_job(resource_defs=resource_defs, config=ops_output_config)

Where the ops_output_config is a conf mapping to generate file paths for each op:

@config_mapping(config_schema={"param_id": int})
def ops_output_config(val):

    conf = load_conf_from_csv(val["param_id"])
    workspace_root = os.path.join(conf.get('run_type'), conf.get('rep_date'), conf.get('nickname'))
    ops_output_config_schema = {}
    for output in ['concat_samples', 'generate_sample1', 'generate_sample2']:
        output_path = os.path.join(workspace_root, 'result', f"{output}.pkl")
        ops_output_config_schema[output] = {'outputs': {'result': {'output_path': output_path}}}

    return {"ops": ops_output_config_schema}

Now the config mapping is not very complex -> is there a way to pass the required ops to the config mapping, so it knows to create records for all of the sample aliases?

I'm providing resources for source assets and normal assets.

source_assets_with_resources = with_resources(
        source_assets,
        resource_defs=resource_defs,
        resource_config_by_key=resource_config_by_key,
    )

    assets_with_resources = with_resources(
        assets,
        resource_defs=resource_defs,
        resource_config_by_key=resource_config_by_key,
    )

This code fails:

Error loading repository location repo.py:dagster._core.errors.DagsterInvalidDefinitionError: Conflicting versions of resource with key 'parquet_io_manager' were provided to different assets. When constructing a job, all resource definitions provided to assets must match by reference equality for a given key.

I'm clearly providing the same resources to different assets (

source_assets

and

assets

lists are not intersecting). This code fails in Dagster:

resource_defs_from_assets = {}
    all_assets: Sequence[Union[AssetsDefinition, SourceAsset]] = [*assets, *source_assets]
    for asset in all_assets:
        for resource_key, resource_def in asset.resource_defs.items():
            if resource_key not in resource_defs_from_assets:
                resource_defs_from_assets[resource_key] = resource_def
            if resource_defs_from_assets[resource_key] != resource_def:
                raise DagsterInvalidDefinitionError(
                    f"Conflicting versions of resource with key '{resource_key}' "
                    "were provided to different assets. When constructing a "
                    "job, all resource definitions provided to assets must "
                    "match by reference equality for a given key."
                )

so for some reason the line

if resource_defs_from_assets[resource_key] != resource_def

evaluates to True. How is this possible if I'm only defining each resource (io_manager) once? I went in with a debugger and apparently these objects are indeed different.

Hi! Couldn’t find how to do this in the documentation. Is there any way to do something like this?

Is it possible to add a graph to a repository dynamically? In other words, can I add a graph and then run it without having to restart the container?

For example, could I make the repository function return a list of jobs that was stored in a database?

Hi everyone, I wanted to query missing partitions using the following GraphQL query in the GraphQL playground:

query test(
  $repositorySelector: RepositorySelector!, 
  $partitionSetName: String!
)
  {
    partitionSetOrError(
      repositorySelector: $repositorySelector
      partitionSetName: $partitionSetName
    ) {
      ... on PartitionSet {
        id
        name
        pipelineName
        partitionsOrError {
          ... on Partitions {
            results {
              name
            }
          }
        }
        partitionStatusesOrError {
          __typename
          ... on PartitionStatuses {
            results {
              id
              partitionName
              runStatus
              runDuration
            }
          }
        }
      }
    }
  }

and I have the following response :

{"error": "Unexpected token '<', \"\n<html><hea\"... is not valid JSON"}

Am I missing something or is it a known problem ? I am using Dagster & Dagit version 1.0.17, deployed on K8S.

Hey folks, is there an option of settings up dagster retries for a pipeline to automatically be re-executed on error? Similar to this article for jobs

trying to get python_logs from code running in dagster graph using the following guide: https://docs.dagster.io/concepts/logging/python-logging using the latest dagit version for some reason i see nothing in the logs after editing dagster.yaml:

python_logs:
  managed_python_loggers:
    - root
    - __main__

The code in the snippet below comes from the documentation: https://docs.dagster.io/concepts/types#defining-a-dagster-type. However it does not work and returns the error below. Is this a problem on my end or in the documentation?

Traceback (most recent call last):
  File "/Users/thomas/miniforge3/envs/datagenerator/lib/python3.10/site-packages/IPython/core/interactiveshell.py", line 3433, in run_code
    exec(code_obj, self.user_global_ns, self.user_ns)
  File "<ipython-input-22-26c8dc208c5a>", line 10, in <module>
    ins={"num": In(EvenDagsterType)},
TypeError: 'list' object is not callable

hey all, is there any matrix chart showing support for Dagster versions VS Kubernetes versions? Or is there any API in the helm chart that isn't compatible given a k8s version?

Looking to clarify my understanding of branch deploys, Do we need an additional dagster_cloud.yaml or should the prod deploy config be sufficient for this?

Hi, I’m trying to work through the dbt tutorial here: https://github.com/dagster-io/dagster/tree/master/examples/tutorial_dbt_dagster#getting-started When I run

dagit

, I get

ModuleNotFoundError: No module named 'dagster_dbt'

$ pip freeze | grep dagster-dbt
dagster-dbt==0.17.6

Any idea what’s going on here?

Hello! We are using dagster-celery to distribute jobs out at the op level by the queue tag. Most of the ops have non-python dependencies (e.g. external tools/executables with varying requirements and dependencies of their own) that we have isolated to separate docker images running the workers. Is there a way to define sort of an interface (possibly similar to a celery signature) for an asset/op/etc. providing the typed inputs and outputs, but having the implementation isolated to the celery worker codebase? According to the docs, the code has to be available in both dagit and the workers. Does the actual compute_fn of an op need to be present in dagit when it's the celery worker executing it?

Can I use DATABRICKS Cluster compute for DAGSTER instance job run ?