I have a resource that acts as a context manager, it has an enter and exit method. If the op fails, will the exit method still be called?

I'm reading through the Dynamic Graphs docs – seems like a super neat approach to mapping over work 👍. One question: does Dagster support limiting the amount of in-flight work? E.g. suppose I have 1MM documents but only want to process 100 at a time (perhaps due to infrastructure constraints) – can I do that? For context, I'm planning on running Hybrid Dagster Cloud, because some of our jobs will require GPU access.

When running a two dagster repositories using gRPC servers, dagit properly shows the assets for both and their connections (one repo calls out SourceAssets with matching keys from the other). But when I try to materialize the assets that are downstream of the other repo, they fail because the upstream IOManager in the other repo is not being used downstream. I would have expected the downstream asset to fail because the IOManager upstream was not declared in the downstream repo but the failure I get is that the pickled file can't be found that the default fs_io_manager uses (instead of the SQL IOManager from the upstream repo being called out as not existing). How do you have upstream IOManagers in other repos get used in downstream assets in their own repo process? I am reading up on this more and went down this path based on this comment in the SourceAsset documentation section: 'You can also re-use assets across repositories by including them as source assets'. But it seems to me that in order for this to work the SourceAsset has to have an IOManager configured on it to properly pull the upstream data even if the upstream asset has an IOManager on it also. I am guessing that there is no synchronization mechanism to pull the configuration from an upstream IOManager down to the downstream IOManager in another repo without manually configuring them. Is this true? And is so is there a recommended way to do this?

Hi all - trying to understand if job "tasks" (or steps maybe they're called) can be run in their own k8s pod, versus each step running in the same k8s pod as the job itself I see: https://docs.dagster.io/_apidocs/libraries/dagster-k8s#dagster_k8s.k8s_job_executor Which might be what I'm after, but want to confirm that a "step" == "op" Edit: but reading this: https://docs.dagster.io/deployment/guides/kubernetes/deploying-with-helm Its not clear whether or not individual steps are ran in their own k8s pod

Hello! Wanted to seek clarification on something. When using an ECS Launcher, the launched jobs will use either the daemon’s image, or the specified image if the

DAGSTER_CURRENT_IMAGE

env var has been defined in the grpc’s container?

is there a recommended approach for generating dagster type(s) from json schema definitions?

Hey dagster team, Running into an error (see screenshots below). The

sce_nol.caiso_discharging_analysis_helper_graph.filter_down_flowgate_analysis_by_number_of_harmers

op is failing and is looking for a dagster file that would be written by a prior op in the job. What's weird is this op fails early on when running the job, but the job run hasn't even gotten to the op yet. E.g. the job is running the ops out of order (see the screenshot where the failed op hasn't had the preceding ops run yet). One thing to note: we're using dynamic outs in the job.

I’m trying to debug configuration of the

GCSComputeLogManager

on a k8s deployment. I have it configured like this in the helm values file:

computeLogManager:
    type: GCSComputeLogManager
    config:
      gcsComputeLogManager:
        bucket: reidab-dagster
        prefix: log/compute/
        localDir: /tmp/k8s-compute-logs/

Authentication is configured for GCS via the

dagster

k8s service account, and I beleive that’s working because the

log/compute

directory was created within my previously-empty GCS bucket. I’m just not seeing any logs get put into the bucket or made available in dagit. (I’m also not sure what the localDir setting is and I’ve tried a few values there. Is that local to the pod being run, or to the runner job, or the user code deployment, or the daemon…?) Any ideas?

Hi, I'm using the

k8s_job_executor

and specifying

tolerations

/

nodeSelector

tags so that pods will run on a specific node group. This works as expected for the

run

pod where I can see that my

tags

were properly applied in the pod's yaml file and the pod runs on a node in the correct node group; however, the

step

pods do not execute on the correct node group and when I examine the

step

pod yaml file the

tolerations

/

nodeSelector

tags are not present. Is this the expected behavior? Is there another way to add tags to individual step pods?

There seems to be a typo/bug in

TableRecord

in the docs: https://docs.dagster.io/_apidocs/ops#dagster.MetadataValue.table It says the signature of a TableRecord is as follows:

TableRecord(code="invalid-data-type", row=2, col="name"}]

Which obviously isn't correct python syntax. It ought to be:

TableRecord(code="invalid-data-type", row=2, col="name")

Upon looking at the actual source code However it still doesn't seem to work. Not sure what the exact bug is, but I get a:

TypeError: TableRecord.__new__() takes 1 positional argument but 2 were given

Hi all. I've written an io manager which returns a

mem_io_manager

or

s3_pickle_io_manager

depending on the config of my custom

get_in_memory_or_s3_io_manager

. This custom io-manager worked fine in dagster

0.14.12

.

@io_manager(
    config_schema=Field(
        Selector(
            {
                "s3": {
                    "s3_bucket": Field(StringSource),
                    "s3_prefix": Field(
                        StringSource, is_required=False, default_value="dagster"
                    ),
                },
                "in_memory": Field(bool, is_required=False, default_value=True),
            },
        ),
        default_value={"in_memory": True},
    ),
)
def get_in_memory_or_s3_io_manager(init_context: InitResourceContext) -> IOManager:
    resource_config = init_context.resource_config

    if "s3" in resource_config:
        return s3_pickle_io_manager(init_context.replace_config(resource_config["s3"]))
    elif resource_config["in_memory"]:
        return mem_io_manager(unbound_context)
    else:
        raise DagsterInvalidConfigError("Error in config for the io manager", [], None)

However, I've just updated my dagster version to

1.0.13

and am now seeing following error:

dagster._check.ParameterCheckError: Param "_" is not a UnboundInitResourceContext. Got <dagster._core.execution.context.init.InitResourceContext object at 0x7f05c06d6970> which is type <class 'dagster._core.execution.context.init.InitResourceContext'>.

Stack Trace:
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/dagster/_core/errors.py", line 184, in user_code_error_boundary
    yield
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/dagster/_core/execution/resources_init.py", line 325, in single_resource_event_generator
    resource_def.resource_fn(context)
  File "/dagflow/dagflow/default_job_configs/custom_resources.py", line 87, in get_in_memory_or_s3_io_manager
    return mem_io_manager(init_context)
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/dagster/_core/definitions/resource_definition.py", line 224, in __call__
    check.opt_inst_param(args[0], context_param_name, UnboundInitResourceContext)
  File "/opt/pysetup/.venv/lib/python3.8/site-packages/dagster/_check/__init__.py", line 682, in opt_inst_param
    raise _param_type_mismatch_exception(obj, ttype, param_name, additional_message)

My questions; • What's wrong with my custom io-manager that this error gets thrown in dagster

1.0.13

? • Are there better patterns to write an io-manager which switches between memory and s3 based on resource config? Thanks !

Hello - we have noticed in our dagster pipeline that there can be long pauses of up to 5 minutes where nothing is running, and nothing is getting prepared, and nothing has errored. Has anyone else seen this type of issue, and know how to begin debugging it? I’ve tried searching the internet but there doesn’t seem to be many hits for this. One suggestion is non-dagster code running behind the scenes for example, but that shouldn’t be applying to us

Hey everyone, I am trying to build a Dagster graph and I'm running into an issue that I am hoping someone can help me with. I am trying to create a pipeline which does three things: 1. Launch an EMR cluster 2. Add a step to that cluster 3. Check whether the step has completed successfully or not. I created the following ops:

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

def launch_emr_cluster() -> Tuple[EmrJobRunner, str]:

emr_config = get_emr_config(

release_label="emr-6.8.0",

cluster_name="Dagster Cluster",

master_node_instance_type="m5.xlarge",

worker_node_instance_type="m5.xlarge",

worker_node_instance_count=2,

ec2_subnet_id="",

bid_price="0.039",

)

emr = EmrJobRunner(

region="eu-central-1", aws_access_key_id="", aws_secret_access_key=""

)

cluster_id = emr.run_job_flow(log=log, cluster_config=emr_config)

<http://log.info|log.info>(cluster_id)

return (emr, cluster_id)

@op

def add_job_step(emr: EmrJobRunner, cluster_id: str) -> Tuple[EmrJobRunner, str, str]:

s3_spark_code_path="<s3://PATH/test-file.py>",

step_defs = get_step_config(s3_spark_code_path=s3_spark_code_path)

step_ids: list = emr.add_job_flow_steps(log=log, cluster_id=cluster_id, step_defs=step_defs)

<http://log.info|log.info>(step_ids)

return (emr, cluster_id, step_ids)

@op

def is_step_complete(emr: EmrJobRunner, cluster_id:str , step_id: str) -> None:

while not(emr.is_step_complete(log=log, cluster_id=cluster_id, emr_step_id=step_id)):

time.sleep(10)

and this is how I build my graph:

@graph

def emr_pipeline():

is_step_complete(add_job_step(launch_emr_cluster(ingest())))

This is how I build my job:

def emr_job():

job = <http://emr_pipeline.to|emr_pipeline.to>_job(

name="daily_etl",

description="Daily ETL job",

partitions_def=DailyPartitionsDefinition(

start_date="2022-11-09", timezone="Europe/Amsterdam"

),

tags={

"dagster-k8s/config": {

"job_spec_config": {"ttl_seconds_after_finished": 300}

}

},

)

job_schedule = build_schedule_from_partitioned_job(

job, default_status=DefaultScheduleStatus.RUNNING

)

return job, job_schedule

However, when I run Dagster I am getting the following error in the UI:

Missing required config entry "ops" at the root. Sample config for missing entry: {'ops': {'add_job_step': {'inputs': {'cluster_id': '...'}}, 'is_step_complete': {'inputs': {'cluster_id': '...', 'step_id': '...'}}}}

I don't understand why Dagster is requesting a runtime configuration when I am passing objects from one op to the next. Can someone help with this? Even pointing to the right documentation would be enough as I've been reading the Ops documentation but haven't been able to resolve it. Thanks in advance!

I think your Algolia service is down:

Hello everyone. I have two assets with configuration possible. If I try to materialize one of the assets (from the asset page) I need to put in the config for the other asset as well. Is there a way to mitigate this (apart from creating a job for each asset)? Minimal reproducible code in the comments and a screenshot

Hi everyone, I’m trying out dagster-cloud as a replacement for our previous data pipeline which relies on pyspark. I created an small poc with dagster that works locally but was now trying to get it running serverless on dagster-cloud. However I’m getting errors with pyspark initialization:

RuntimeError: Java gateway process exited before sending its port number

. Is this a limitation of dagster-cloud or do I need to setup some other things to make it work?

Hi! Asset sensors are great, but is there a way to create a sensor for prefixes? Like the docs say you can use keys:

@asset_sensor(
    asset_key=AssetKey("my_awesome_job"),

I'm using Asset Materializations, and would like to listen for anything on

foobar_

. edit: seems i've hit https://github.com/dagster-io/dagster/issues/4582

Hi, I have been using dagster on k8s and able to run my jobs on k8s run launcher. However the sensor I have based on job success status is unable to start due to "unable to load environment varaible" envvars. all jobs, sensors are set to run under a reposiotory with execute def = k8s. any idea which config change is required to sensors to pick up env vars. Note: I am using helm chanrt/values.yaml

hello, I’m trying to run 2 different repositories using dagit with workspace.yaml. I’ve defined the following workspace file:

load_from:
  - python_file:
      relative_path: my/path/subpackage1/repository.py
  - python_file:
      relative_path: my/path/subpackage2/repository.py

and I get the following error:

dagster._check.CheckError: Invariant failed. Description: Cannot have multiple locations with the same name, got multiple "repository.py"

Then, if I rename one of the two

repository.py

files with another name then the errors disappears but

dagit

looks stuck (the server is not spawning and if I try to abort with ctrl-C it doesn’t stop). I’m using

dagit 1.0.17

Any suggestions? Thanks

hello, new dagster version has a migration which does CREATE INDEX CONCURRENTLY , which basically claims exclusive access over the whole DB. https://www.postgresql.org/message-id/MWHPR20MB1421BF01C05147053F6F64CAA19D0%40MWHPR20MB1421.namprd20.prod.outlook.com so for me this never completes

Hi all, is it possible to materialise multiple partitioned assets in one go within python and to not recompute assets which do not depend on the partitioned asset?

from dagster import StaticPartitionsDefinition, materialize_to_memory, asset


@asset
def independent_asset():
    return 1 + 2


@asset(partitions_def=StaticPartitionsDefinition(["1", "2"]))
def partitioned(context):
    return context.partition_key


@asset
def top_level_asset(partitioned, independent_asset):
    return partitioned, independent_asset


materialize_to_memory([top_level_asset, partitioned, independent_asset], partition_key=["1", "2"])

Good afternoon! I've got a small question regarding the visibility of default/hardcoded configuration of Ops (https://docs.dagster.io/concepts/ops-jobs-graphs/jobs#hardcoded-configuration). We're using some hardcoded config for Ops in (Graph backed) Assets and we expected the values to be shown on the sidebar of the Op, but only the schema is shown not the hardcoded values. Attached is a screenshot from the Op from the Graph, but the Job shows the same. I'm aware you can see (some https://github.com/dagster-io/dagster/issues/7227) of this via the launchpad, which I understand is necessary when you want to manually create/trigger a Run, but it seems like it should be possible to just view the config when just viewing the Asset/Graph/Job and we expected this to be shown where the schema is currently shown.

Hello! I’m using run retries for some of my jobs. In my DAG, I’m determining if the current run is a retried run by looking for the

parent-run-id

in the op context object. However, that field is set for both automatic run retries and user-initiated retries in the Dagster UI. Is there a programmatic way to differentiate between automatic retries vs user retries?

Hello, is there a way to link the historical runs of a job after is was renamed?

Hey! is it possible to access the metadata provided by

Outputs

and

DynamicOutputs

in an

IOManager

at runtime? Or only the metadata provided by the

Out

class at the

op

definition?

hey yall - if we have a k8s based dagster deployment that reads its workspace config from a ConfigMap will it automatically pick up new entries added? (assuming we click "reload all" in the ui)

I have an asset that depends on many upstream assets. Is there a way to programmatically define upstream assets instead of writing all of the upstream asset names as args?

Hello, could someone help me out with a problem? I'm having trouble trying to figure out how to go about discovering what the actual complaint is with my computer overlords. Error Message:

botocore.errorfactory.InvalidParameterException: An error occurred (InvalidParameterException) when calling the RunTask operation: launch type should be one of [EC2,FARGATE,EXTERNAL]

dagster.yaml

run_launcher:

module: "dagster_aws.ecs"

class: "EcsRunLauncher"

config:

container_name: "my-dagster"

run_task_kwargs:

launchType: "Fargate"

I'm thinking that I didn't configure my Dagster to use Fargate correctly.

Hi Team, What is the maximum limit of the pipelines in a Repo ?

Hello everybody, I would like to ask a question about the best way to deploy 2 versions of a pipeline at the same time. I will post the details and some pictures in the thread of this discussion.