Is there a way to validate the dagster usercode repo? something like

dagster validate --path

? I'm thinking CI/CD

Hi, I'd like to send a small PR to dagster. Do I make it against master? the contributing guide doesn't specify

Hello all. Can't find UI to re-execute all selected jobs in Dagit Runs. Only "Terminate..." and "Delete" commands are available in Actions. Is there another way to do this in the UI?

Hello, I'm using Dagster within a HTTP REST endpoint. Basically I'm executing a pipeline using the execute_pipeline api. In my case there were 3 independent solids which are not dependent on any other solids yet they run in sequentially, I noticed them running in parallel when I run it using the dagit or CLI. I found the constraints around multiprocess in dagster python api and did applied the execution configuration in run_config. It is working fine. But can I do that without using multiprocess? like using threads or coroutines? Reason is I want to execute them in async without blocking and also parallelly for faster execution. I thinking of trying the execute_pipeline_iterator for async but I couldn't fine any clear documentation for it. TIA

hey, i just ran into a weird issue when running a job locally through dagit where dagster seemingly can’t find the temporary directory being used for an op. any ideas?

How can a graph have multiple outputs? The documentation says

To map multiple outputs, return a dictionary from the composition function.

but I don't quite understand it

I have a resource that I was using to collect information across multiple ops for an in_process_executor job which works but when I switch to the multiprocess execution, it's unable to retain its state (seems like it's creating new instance of it per op it's used). Is there a good way of handling this?

Just wanted to flag this guy https://github.com/dagster-io/dagster/issues/6057 as it was brought up right before the holidays. If the new partition config api is expected to fully cover its predecessor atm, I do believe

{frequency}_partition_configs

apis are not fully there yet. https://docs.dagster.io/guides/dagster/graph_job_op#partition-schedules

I had an op retry with a

STEP_UP_FOR_RETRY

event - but my op doesn’t have a retry policy defined. Is there a default retry set somewhere? Where can I adjust this?

Hi All -- where can I find information about the syntax that is supported when filtering ops in the Dagit UI?

I’m finding the config/resource system confusing. I want to define a job like the following:

@job
def do_load():
  data = get_data()
  processed = process_data(data)

  if environment == 'dev':
    # do dev-specific operation
    # (but don't load_data until that's done)

  load_data(processed)

and be able to have each op take settings for the appropriate environment, and then define “configured” versions of the job like

do_load_dev

,

do_load_staging

,

do_load_prod

that can have the ops use different settings per environment. Currently I have a yaml config like the following:

resources:
  environment:
    config:
      prod:
        user: "prod_user"
        url: "https://.../"
      staging:
        user: "staging_user"
        url: "https://.../"

I can read that resource with:

@op(required_resource_keys={"environment"})
def do_thing_with_resource(context):
    env = context.resources.environment
    print(f"Got env: {env}")


@job(resource_defs={"environment": make_values_resource()})
def run_load():
    do_thing_with_resource()

but what’s the right way to preconfigure that per-environment and be able to access the specific environment settings in each op? Also, what’s the right way to set environment variables for code or shell commands (

create_shell_command_op

) in an @op? Thanks!

Hi, what is the recommended way to specify external python libraries (dependencies) when running on pyspark?

Hi all, does anyone know how to change host & port on dagit?

Hi all! Is commercial support available for Dagster?

Hi All, My requirement is to generate ops dynamically and all generated ops should be linked serially. I tried dynamic mapping and collect but that runs the ops parallelly . for example: assume list = [('title1', 'query1'), ('title2', 'query2'), ('title3', 'query3')] this list can be fed to op generator which create first op named title1 and do some stuff with supplied args (query1) and return success or fail then create 2nd and repeat for entire list.

Hi folks, I’m using dagster-k8s, and aim to pass

env_secrets

to job as I spotted https://docs.dagster.io/_modules/dagster_k8s/executor#k8s_job_executor

env_secrets=run_launcher.env_secrets + (exc_cfg.get("env_secrets") or []),

Hence, my code is following

download_host_fe_job_stg = download.to_job(
    name=f'{NAME}_stg_fe',
    resource_defs={
        "io_manager": s3_pickle_io_manager,
        "s3": s3_resource,
    },
    config=partitioned_config,
    tags={"dagster-k8s/config": {
            "job_config": {
                "env_secrets": [
                    "dagster-gcp-secret-fe"
                ]
            }
    }}
)

but I got error

TypeError: __init__() got an unexpected keyword argument 'env_secrets'
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/implementation/utils.py", line 34, in _fn
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/implementation/execution/launch_execution.py", line 16, in launch_pipeline_execution
    return _launch_pipeline_execution(graphene_info, execution_params)
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/implementation/execution/launch_execution.py", line 50, in _launch_pipeline_execution
    run = do_launch(graphene_info, execution_params, is_reexecuted)
  File "/usr/local/lib/python3.7/site-packages/dagster_graphql/implementation/execution/launch_execution.py", line 38, in do_launch
    workspace=graphene_info.context,
  File "/usr/local/lib/python3.7/site-packages/dagster/core/instance/__init__.py", line 1455, in submit_run
    SubmitRunContext(run, workspace=workspace)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/run_coordinator/default_run_coordinator.py", line 32, in submit_run
    self._instance.launch_run(pipeline_run.run_id, context.workspace)
  File "/usr/local/lib/python3.7/site-packages/dagster/core/instance/__init__.py", line 1519, in launch_run
    self._run_launcher.launch_run(LaunchRunContext(pipeline_run=run, workspace=workspace))
  File "/usr/local/lib/python3.7/site-packages/dagster_k8s/launcher.py", line 312, in launch_run
    user_defined_k8s_config=user_defined_k8s_config,
  File "/usr/local/lib/python3.7/site-packages/dagster_k8s/job.py", line 649, in construct_dagster_k8s_job
    **user_defined_k8s_config.job_config,

Please help me to correct configs for a k8s job.

I hit this exception

botocore.exceptions.EndpointConnectionError: Could not connect to the endpoint URL: "<https://behold-pipelines.s3.eu-west-2.amazonaws.com/?list-type=2&prefix=dagster%2Fstorage%2Fb42b4c9c-b930-4f6d-8a1e-58ea43d01361%2Ffetch_instances_info%5B12%5D%2Fresult&encoding-type=url>"
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_plan.py", line 193, in _dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 326, in core_dagster_event_sequence_for_step
    for evt in _type_check_and_store_output(step_context, user_event, input_lineage):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 380, in _type_check_and_store_output
    for evt in _store_output(step_context, step_output_handle, output, input_lineage):
  File "/usr/local/lib/python3.7/site-packages/dagster/core/execution/plan/execute_step.py", line 488, in _store_output
    handle_output_res = output_manager.handle_output(output_context, output.value)
  File "/usr/local/lib/python3.7/site-packages/dagster_aws/s3/io_manager.py", line 68, in handle_output
    if self._has_object(key):
  File "/usr/local/lib/python3.7/site-packages/dagster_aws/s3/io_manager.py", line 50, in _has_object
    key_count = self.s3.list_objects_v2(Bucket=self.bucket, Prefix=key)["KeyCount"]
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 386, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 692, in _make_api_call
    operation_model, request_dict, request_context)
  File "/usr/local/lib/python3.7/site-packages/botocore/client.py", line 711, in _make_request
    return self._endpoint.make_request(operation_model, request_dict)
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 102, in make_request
    return self._send_request(request_dict, operation_model)
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 156, in _send_request
    raise exception
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 200, in _do_get_response
    http_response = self._send(request)
  File "/usr/local/lib/python3.7/site-packages/botocore/endpoint.py", line 269, in _send
    return self.http_session.send(request)
  File "/usr/local/lib/python3.7/site-packages/botocore/httpsession.py", line 352, in send
    raise EndpointConnectionError(endpoint_url=request.url, error=e)

I'm assuming the connection to S3 was lost temporarily, I was going to add RetryRequested within the op but because this S3 load is handled by the io_manager I don't think that would increase the robustness at all. How could I make the io_manager try again for this op?

Hi All, Just a quick question, We all know that we can a schedule a pipeline in Dagster. I know we can control the execution of specific solid in a pipeline via Dagit manually, but is it possible to schedule a specific solid and not the whole pipeline ?

I frequently find myself wanting to pass some metadata between ops along with the actual data. For example, op A may create a dataframe and op B transforms that dataframe. If this is a ETL workflow that is going to persist data to a data warehouse I may want to include a tablename with that dataframe. Right now I create a dict and pass that between tasks. I was curious if there was a way for op A to output an Output that includes the value and metadata and then for op B to read the metadata along with the value.

{
    "table_name": "some_new_file.json",
    "value": df
}

Hi! I'm trying to use multiple AWS profiles in my pipelines. Here is the relevant part of my values for dagster-user-deployments helm chart:

deployments:
  - ...
    env:
      AWS_SHARED_CREDENTIALS_FILE: /etc/aws/credentials
    volumes:
      - name: aws-credentials
        secret:
          secretName: aws-credentials
    volumeMounts:
      - name: aws-credentials
        mountPath: /etc/aws
        readOnly: true

I've deployed the chart and tested the configuration from inside the pod (with python shell and boto3): it works. But when I run the job that have to save something to a bucket with Dagit I face the issue

Initialization of resources [io_manager, s3] failed.
botocore.exceptions.ProfileNotFound: The config profile (myprofile) could not be found

As I understand I have to configure something else. Could you help me please?

I've also faced another problem: Dagit doesn't reload

@repository

when I build new user-deployments image and upgrade the helm chart. But the pod is updated, so Dagit can't run it, but it doesn't see the changes. I have to uninstall and install the chart again to make it work. I'm using "latest" tag on my image. Could that be a reason?

Hi folks, is there a way to merge partitioned configs and other configs? Example, I have a partition like below

@daily_partitioned_config(start_date=datetime(2021, 11, 1))
def partitioned_config(start: datetime, end: datetime):
    return {
        'ops': {
            'my_func': {
                'config': {
                    'date': start.strftime("%Y%m%d")
                }
            }
        }
    }

aims to merge with io_manager’s configs

from dagster.utils import merge_dicts

download.to_job(
    resource_defs={
        "io_manager": s3_pickle_io_manager,
        "s3": s3_resource,
    },
    config=merge_dicts(partitioned_config, {
        'resources': {
            'io_manager': {
                'config': {
                    's3_bucket': "dagster-test",
                    's3_prefix': "dagster"
                }
            }
        }
    })
)

Hi! I'm trying to update from Dagster 0.11.13 to 0.13.13 but our pipelines with DynamicOutputs no longer run. The solid that yields DynamicOutputs runs successfully but when we map that to our

get_metric_timelines

solid we're now getting the attached error message. We're using the k8s_job_executor and previously this would create a new pod for each dynamic output. I'm having trouble figuring out what causes this

Is there a release coming out today?

#dagster-support @daniel can someone help with troubleshooting an issue with the ephemeral run pods in a k8 cluster missing a kubernetes secret? you can see the secrets present on the cluster, and the

runLauncher.config.k8sRunLauncher.envSecrets

in the helm chart referencing them, but the pods aren’t spun up with the secrets (in my .py @op code i use

os.getenv('the_actual_secret_name')

to troubleshoot and confirm that they are missing) 🙏🏻 thank you box

Hello everyone ! I am trying to couple ArgoCD with dagster but I amstuck in the image deployment of user code. When I try to override in ArgoCD values.yaml it pulls the wrongversion of the image (0.0.1-dev instead of 0.13.13). It is in the middle of both technologies so it might be related to the way I ArgoCD ? I find where is the problem and solves it manually but now it's how I do it automatically. Here is the issue. At the end I may publish a blog post on how to do it. https://github.com/dagster-io/dagster/issues/3099

Hello all. Dear team, are there any plans to add "smart" filters to the Run UI widget like

tag1 > 1 and (tag2 <= "2022-01-14" or tag 3 == "test value")

?

HI, I have created an ETL pipeline using Dagster. It has three Jobs - Job1 (reads data from file and writes to Kafka. Each message has an associated name), Job2 (reads from Kafka carries out transform A and writes processed data to Kafka) Job 3 (reads from Kafka and writes to Postgresql). When data is read from the file, there is a requirement to compare it to the last message update from the same name. This information can be taken from the Postgresql table. However, since the data rate is very high, it can't be done in real-time. So the pipeline reads the data from Postgres on the first run and creates a list of latest update for each name. Thereafter, it is maintained and updated in memory. Thus everytime Job1 runs, it uses this memory object for comparison and updates it based on input from file. So briefly, requirement is - 1. Read and create dict/dataframe object from Postgresql on first Job run. 2. Update dict/dataframe during each Job run and save. 3. Updated dict/dataframe should be available to every subsequent Job run. What would be the best method for this? Would be very grateful for any help and advice. Thanks.

Hey everyone, quick question: will I be able to use dagster to create graphs at runtime? Let's say I have an ETL graph defined in a document somewhere (e.g. JSON) and can map the graph nodes to Python operators / dagster.ops, is it possible to compose the graph at runtime? I'd like to avoid declaratively defining the graph in code, but rather do it at runtime. I guess I could call dagster's decorators directly, but then I'm currently unsure how to dynamically reload / add graphs to a repository and what API I'd use to launch those, apart from

execute_in_process

, which I don't think does everything dagster would be able to do if used right. Right now, I'm under the impression dagster seems to be built around the declarative assumption and I don't know if there is an API that allows me to use dagster's features like repos, mapping execution to containers etc. given my use case above - any pointers are greatly appreciated, thanks

Hi, I'm experiencing a bug when trying to write an

ALS

model from

pyspark.ml.recommendation

to S3 and reading it back in if this takes place within a dynamically executed graph (i.e. via dynamic mapping). I wrote a custom IO manager using the pattern

f's3a://{self.s3_bucket}/{key}'

as

_uri_for_key

similar to the one currently implemented in the

PickledObjectS3IOManager

. As the step identifiers for the dynamically generated steps contain square brackets [ and ] these are included in the S3 uri when an object is written. Even though I can clearly see the model was saved to this path in S3, I'm getting an error when the downstream op tries to load the model, something like:

py4j.protocol.Py4JJavaError: An error occurred while calling o44.load.
: org.apache.hadoop.mapred.InvalidInputException: Input Pattern s3a://....fit_model[...]/model/metadata matches 0 files

When I replace/remove the square brackets from

_uri_for_key

this works fine:

f's3a://{self.s3_bucket}/{key}'.replace('[', '_').replace(']', '')

It seems that technically

_uri_for_key

is only used for debug logs in the

PickledObjectS3IOManager

and the writing/reading occurs via

upload_fileobj

and

pickle.loads

without actually using this key. I can imagine that the same error could occur with the

PickledObjectS3IOManager

and thought I'd point it out here, in case this hasn't been tested yet. Moreover, square brackets are among the characters to avoid in S3 object keys according to https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html On the other hand, maybe the error is specific to my particular use case if: 1. the resulting uri from

upload_fileobj

is actually different from the one generated when using

upload_fileobj

, which would mean that the debug log message should be corrected 2. maybe the error only occurs when trying to read a folder which is the case for the

ALS

model (see error message) but not for pickled objects 3. maybe the error only occurs within Py4J