what's the recommended strategy to build a dagit container with a single workspace and a single repo? My current setup is •

DAGSTER_HOME=/opt/dagster/dagster_home

. Copy the dagster.yml there (works just fine) • workdir is

/opt/dagster/app

. I copy all my files there, including

/opt/dagster/app/my_project/repository.py

and the

workspace.yaml

in the root • my workspace contains

python_file: relative_path: my_project/repository.py

. This loads correctly when I run

dagit

locally but not in docker (

module my_project was not found

) What's the best practice there to make sure that the import works?

Are there any plans to support

Ray

? (https://github.com/ray-project/ray) and for my understanding was there any reason to choose

Dask

over

Ray

for multi-processing?

Hey ! I want to mock a resource I created, so I used magic mock as following:

from unittest.mock import MagicMock
mocked_http = http_async_downloader()
mocked_http.fetch_all = MagicMock(return_value=["4 /n hello"])


@job(resource_defs={"http": mocked_http})
def mocked_job():
    print_htmls(download_urls(convert_cids_to_urls()))


if __name__ == "__main__":
    # upload_pubchem_3d_conformers_dev_job.execute_in_process()
    mocked_job.execute_in_process(run_config=config_from_files([file_relative_path(__file__, 'config.yaml')]))

but I got the following exception:

dagster.check.CheckError: Value in dictionary mismatches expected type for key http. Expected value of type <class 'dagster.core.definitions.resource_definition.ResourceDefinition'>. Got value <resources.aka_http.HttpAsyncDownloader object at 0x7fd8b05cc390> of type <class 'resources.aka_http.HttpAsyncDownloader'>.

what am I missing ?

Hey Guys, our dagster db grows extremely large, I've looked into it and it's the json events that take a lot of space. Is there any option to retain a 30 day amount or roll it over into an archive?

Hey all, Is there a way to create generic sensor (using job name from email config) to trigger the job from any of the existing repositories. existing repos. <dagster_repo>: Generic sensors repo_1>: Job1, job2,..job10 repo_2>: Job11, job12,..job20

Question concerning deployments. Would the proper pattern (for docker-compose, soon to be kubernetes) docker-compose.yaml, dagster.yaml and .env actually physically residing on the servers for each of the environments and the pipeline code pulled from a docker repository?

We have been trying to deploy dagster on helm for a few hours now and we are hitting this problem:

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError) connection to server at "<http://core-shared-tools.cktverwlkjvp.us-east-1.rds.amazonaws.com|core-shared-tools.cktverwlkjvp.us-east-1.rds.amazonaws.com>" (10.20.177.69), port 5432 failed: FATAL:  password authentication failed for user "dagster"

42
connection to server at "<http://core-shared-tools.cktverwlkjvp.us-east-1.rds.amazonaws.com|core-shared-tools.cktverwlkjvp.us-east-1.rds.amazonaws.com>" (10.20.177.69), port 5432 failed: FATAL:  password authentication failed for user "dagster"

I know what you are thinking, have you verified that the password is correct? We have tried so many combinations of setting this password. autogenerated, from vault, plain text in the values.yaml file. With an external postgres instance, without. And it always says it is incorrect. I was wondering if anyone has hit a similar issue? This is the values.yaml file we have now:

There is an example here https://docs.dagster.io/guides/dagster/software-defined-assets which shows how to write an IOManager that can handle_output() and load_input() for Panda dataframes. In the load_input() function, the code uses 'context.dagster_type.typing_type' to derive the type of the dataframe to input. However, this field is never set to anything in the version that I am using (0.14.3). Any thoughts on what field should be used instead to derive the input type to load into? def load_input(self, context) -> Union[PandasDF, SparkDF]: if context.dagster_type.typing_type == PandasDF: NOTE: I am not using software defined assets in my code yet.

I can't seem to figure out what is wrong with my schedule. I keep getting the below error.

dagster.daemon.SchedulerDaemon - WARNING - Schedule stock_market_data_job_schedule was started from a location StockData.py that can no longer be found in the workspace, or has metadata that has changed since the schedule was started. You can turn off this schedule in the Dagit UI from the Status tab.

I confirmed in dagit that Daemon is running, clock symbol is green next to my job & I believe that I am using the same workspace.yaml.

Logging from a sensor I'd like to be able to log from a sensor using the logging as defined under

python_logs

in

dagster.yaml

. There is no logger available on

SensorEvaluationContext

. I thought I could get a logger using

dagster.get_dagster_logger

and use that in a sensor, but I tried that and it doesn't seem like the log messages are captured anywhere. My key use case is getting notified if a sensor fails primarily so that I can be proactively notified if a job that should run (that a sensor would have kicked off) doesn't. I looked into using a

failure_hook

, but it doesn't seem to be designed to work on sensors. Any tips? https://docs.dagster.io/0.13.19/_apidocs/utilities#dagster.get_dagster_logger

Hi, I had a scheduler to run a test job run every 15 seconds. After confirming it worked, the scheduler was then deactivated (switched off using dagit ui) but the task still runs every 15 seconds. I closed the dagster-daemon and dagit to stop that, but it starts again as soon as I restart dagster-daemon, and it repeats the grpc error message below. How can I hard reset this so that this weird behavior stops? The

Fixed Fan-In with Order-Only Dependencies I'm trying to create a graph with data and order-based dependencies (see attached diagram).

op A

produces some data that ops in

graph C

depend on (data dependency).

op B1

,

op B2

,

op B3

, all need to complete before

graph C

should start (order-based dependency).

op B1

,

op B2

,

op B3

execution order is not important. Can anyone suggest a way to build this graph please?

Hey everyone, I see that there is a new release 0.14.3 but I don't see it published anywhere, just a new git tag. Is it some kind of pre-release?

I'm trying to configure a daily partitionned job without prior knowledge of the start partition date (the first partition date is a timestamp included in the filename I'm supposed to process). What would be the best dagsterish way to proceed in this case?

@op
def op_great_expectations(context, df: DataFrame):
    
    ge = ge_validation_op_factory(
        name='op_great_expecations',
        datasource_name='test_data',
        suite_name='basic.warning',
    )(context=build_solid_context(
        resources={'ge_data_context': ge_data_context},
        config={
            'resources': {
                'ge_data_context': {
                    'resource_config': {'ge_root_dir': file_relative_path(__file__, '../kh_great_expectations')}
                }
            }}
    ))
    res = ge(df)

    if res['success']:
        return df
    return ValueError

Hi all, we are trying to integrate ge_validation_op_factory to our dagster job. We are facing the below error. great_expectations.exceptions.exceptions.ConfigNotFoundError: Error: No great_expectations directory was found here! - Please check that you are in the correct directory or have specified the correct directory. - If you have never run Great Expectations in this project, please run

great_expectations init

to get started. we are passing the correct path, but still facing the error. please do let us know if there are ways to resolve this error. Note: 'df' would be the output from the previous op

Hey team, is it possible to add custom event attributes to datadog traces in Dagster?

Just out of curiosity: Was anyone able to attach to a debugger (e.g. debugpy) to a docker container running dagster? I don't know where to run debugpy - on dagit or on the pipeline docker container?

Hi, i'm trying to use dagster to run airbyte for the API shopify and I found that for the endpoints customers and pages have the error in the file. Does someone know if how I can fix this? Thanks.

Hi, I just started learning dagster and want to get a sense of what a minimal scheduler looks like. I have this simple pipeline with the command to run it at the end:

# orchestration.py

from dagster import job, op
from os import system

@op
def test():
  system("ls")
  
@job
def run_pipeline():
  test()
    
if __name__ == "__main__":
  result = run_pipeline.execute_in_process()
    
# dagster job execute -f orchestration.py

This is my framework for extending it to run every 5 minutes. What am I missing from the script?

# orchestration.py

from dagster import job, op
from os import system

@op
def test():
  system("ls")
  
@job
def run_pipeline():
  test()
    
@schedule(
    cron_schedule = "*/5 * * * *",
    job = run_pipeline,
    execution_timezone="US/Central",
)
def run_pipeline_schedule():
  pass
    
if __name__ == "__main__":
  pass
    
# dagster schedule start
# start dagster-daemon?

Hi all! We are running a quite large daily pipeline that at some point needs to run ~300 high-resource, long-ish running ops in parallel that train an ML model each. To achieve this, we are executing each of the steps as a k8s Job via the celery kubernetes job executor, and autoscaling the kubernetes cluster as needed with karpenter. However, our pipeline also includes a bunch of faster-running ops to prepare data for the model training. Many of these smaller ops take of the order of ten second, and in our current pipeline there are ~20 of these per model-training op. We have found that when running all of this in parallel, the kubernetes API server and its etcd store are overwhelmed because of the high rate of Job and Pod resource creation and destruction across the nodes in the cluster (even using multiple masters or managed k8s control panes like EKS or GKE), and many op executions end up failing. What options do we have to achieve stability in execution? The only mitigation that we can see right now is bundling all ops in data preparation so it is a single op per model. However, this would make it harder to reuse some of the internal op results across models (e.g., some of the test split within these ops gets used after all the models are trained to compute overall test statistics). Another option might be splitting the pipeline into different jobs that communicate via asset sensors. The data prep pipeline would run as a single pod, whereas the model training would run with a pod per model. However, this degrades developer experience, as there is more cognitive overhead in connecting the outputs of one run of the data prep pipeline with the same day's run of the model training pipeline compared to a single pipeline. In an ideal world, we would like to be able to group ops to be executed within the same pod (i.e., as if we were using the

K8sRunLauncher

with multiprocessing executor) but then execute some ops in their own pod (i.e., as if we were using the

k8s_job_executor

). An option to define different executor for a

@graph

within a pipeline would be awesome in terms of flexibility. Has this option been considered?

Hi Team, I'm trying to use

build_reconstructable_pipeline

which takes module name and function name as string to do the dynamic imports later using importlib. It was working fine within sub directories and modules in Dagster 0.13.0 but now throwing below error in 0.14.3 It is pythonic execution

dagster.core.errors.DagsterImportError: Encountered ImportError: `No module named 'builder'` while importing module builder. If relying on the working directory to resolve modules, 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.

set the channel topic: ⚠️ The Dagster core team is at an offsite event and will have limited bandwidth Mar 8 - Mar 10. Thank you for your patience! 🙋 Ask your question after checking out our documentation! https://docs.dagster.io/

Hi, I have an op that return dataframe. I ran the op (read_db_op() ) from the @graph by .to_job.

@graph
def read_db_graph():
    read_db_op()


sqlalchemy_cheese_job = read_db_graph.to_job(
    resource_defs={"db_resource": sqlalchemy_resource},
    config=config_from_files([file_relative_path(__file__, "config_cheesedb.yaml")]))

then - I want to get the data frame.

result_df = sqlalchemy_cheese_job.execute_in_process()

but I got an object : dagster.core.execution.execute_in_process_result.ExecuteInProcessResult object what did I do wrong ?

Hey all - I am missing the idiomatic way to get secrets into ops. Have been through the docs but can't see it. I mean things like AWS creds as well as simple keys that I would normally surface from an env variable or retrieve from Vault. Anyone got any pointers?

I may have missed this in instructions but is there a way to define in code that you want a sensor turned on immediately? I had created a sensor to send PagerDuty alerts for any jobs that fail, but I failed to go to the UI to turn it on. It took me a while to figure out that was the problem. I’d like to avoid that in the future for other sensors by controlling that in code if possible. Is there a way to do that?

Upgrade to 0.14.3: Cannot import name 'calculate_delay' from 'dagster.core.definitions.policy' I've just upgraded from 0.13.19 to 0.14.3. I'm getting the following error when launching a task the the lanchpad.

dagster.core.errors.DagsterLaunchFailedError: Error during RPC setup for executing run: ImportError: cannot import name 'calculate_delay' from 'dagster.core.definitions.policy' (/opt/venv/lib/python3.9/site-packages/dagster/core/definitions/policy.py)

File "/opt/venv/lib/python3.9/site-packages/dagster/core/instance/__init__.py", line 1575, in launch_run self._run_launcher.launch_run(LaunchRunContext(pipeline_run=run, workspace=workspace)) File "/opt/venv/lib/python3.9/site-packages/dagster/core/launcher/default_run_launcher.py", line 106, in launch_run raise (

Any tips on what I might have done wrong?

Hi All, I want to create a dynamic asset from a function. Is that possible? Here is my code example, when I run it using dagit -f, I can get an error saying Input asset '"lst"' for asset 'generate_assets' is not produced by any of the provided asset ops and is not one of the provided sources

@asset(required_resource_keys={'coinmetrics_client'},
       io_manager_key='io_manager', dagster_type=AssetMetrics,
       metadata={'dataset_id': CM_ASSETS_LIST_DSID})
def generate_assets(context, lst):
    # This function takes a key and generates a dataframe which needs to be stored as an asset.
    return asset_metric_df

def generate_subtasks(df: pd.DataFrame):
    for row in df.iter_tuples():
        yield(row[0])

def master_asset(df: pd.DataFrame):
    # For each row in df, generate an asset
    generate_subtasks(df).map(lambda x: generate_assets(x))

hello, in dagster_k8s_executor, the env_vars is not passed to the Job. so i can not pass environment variables to the individual ops when i use the executor. For the k8s_run_launcher it works, its just for the executor it doesn't

Hi, does someone have a complete example on how to integrate slack to send messages on failure or success, because i dont understand the one in the documentation. Thanks

Hey all, starting to look into what deployments for dagster look like, we are an azure cloud based company and we were looking into what permissions certain tooling of dagster will require. Looking into the

compute_logs

looks like one of the avialble resources in the AzureBlobComputeLogManager. Looking at the required inputs looks like it requires a storage account key. Is it possible to hook this up utilizing managed identity / service principal instead? If not is compute_logs also something that we can create our own custom solutions?