Hey! I'm currently building a pipeline in which I have an

op

that access the

run_config

in it (using

context.run_config

). When I tried to develop tests for this

op

, I found out that I cannot set the

run_config

in the

build_op_context

. Is that correct? If so, the best way to test it is doing something like this?

"""file: test_op.py"""

@graph
def new_job_to_test_op(inputs):
	op_to_be_tested(inputs)

def test_op():
	run_config = ...
	testing_op = job_to_test_op.to_job()
	testing_op.execute_in_process(run_config=run_config)

Does the

ScheduleDefinition

solid_selection

parameter work for op names? If any ops belong to nested graphs, should we include the graph name for the

solid_selection

? I'm having trouble getting this to work - Whenever I put a list of op names inside

solid_selection

(with or without the nested graph names), the schedule's launched runs included all of the ops, not just the selected ones.

Is there a way to check the peak memory usage of an op once it is done executing?

OK - I'm missing a trick. I have defined an asset that I wish to materialise to S3 via the

s3_pickle_asset_io_manager

. I've configured it as per the doc page and it's popped up in dagit but when I try to materialise it, it complains about missing config for the s3 io manager (fair enough, I haven't given it any). Where on earth an I supposed to provide that though?

There's a specific example using this IO manager in the docs that doesn't mention this config anywhere

with build_op_context(
    resources={'ge_data_context': ge_data_context},
    resources_config={
            'ge_data_context': {

                'config': {
                    'ge_root_dir': file_relative_path(
                        __file__,
                        '../great_expectations'
                    )
                }
            }
        }
) as context:
    ge = ge_validation_op_factory(
        name='great_expecations',
        datasource_name='test_data',
        suite_name='target.warning',
    )(context=context)

    res = ge(df)

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

Hi all, I have tried invoking ge_op_validation_factory using build_op_context as a context manager, but facing the below error """ No value provided for required input "dataset" """" please do let me know where can i provide the dataset

Hi team, I am extract data in long time. With such a large amount of time, it cause highload our system so I will divide by batch(day by day). But I don’t want config for every run playground. please let me know to pass time range to pipeline and after that they will auto split by day by day. thanks

Hi guys, I've developped several pipelines with resources that need to be configurated. I can run them fine in dagit by adding the configuration in the launchpad, but it doesn't seem to be used when run on schedule, and I haven't found a way to specify the configuration to the schedule. What's the recommended way to provide configuration to a schedule ?

Hi, I'm having problems making a schedule: I made the following schedule like in the docs: basic_schedule = ScheduleDefinition(job=job_orchestrator, cron_schedule="57 11 * * *") I have created the folder dagster_home and run the dagster deamon I used the comand: dagster schedule start and it didn't work I put the following in diferent yaml files: - scheduler: - module: dagster.core.scheduler - class: DagsterDaemonScheduler Does someone have an example with all of the things I have to do to run an schedule? Thanks

In airflow I could write templated queries with jinja syntax

{{ }}

and I could also see the rendered SQL, is something like that possible with dagster ?

Hello! Does dagster support versioning of data assets that goes through the pipeline?

Hi I am working on a pipeline to read data from a source db and write it to our data warehouse. Right now I have it working with ops defined for each table and the destination table is set in the metadata for the io_manager in the Out but I would ideally like to just have one op where I can pass it a list of tables and loop through them. I am not sure how I would update the destination table if I do that though. This is an example of one of the ops:

@op(

required_resource_keys={'warehouse', 'snowflake_db'},

out=Out(

io_manager_key="snowflake_io_manager",

metadata={"table": "DAGSTER.CHATS", "partitioned": False},

))

def load_data_chats(context, incremental=True, table='chats'):

if incremental:

max_update_date = str(context.resources.snowflake_db.get_max_updated_at(table=table,schema='DAGSTER'))

return context.resources.warehouse.incremental_load_from_table(table, max_update_date)

else:

return context.resources.warehouse.full_load_from_table(table)

Hi all, Is it possible to use

if __name__ == "__main__":

with a scheduled job? I think the docs show a way of using that syntax with a job, but only when executing a job outside of a schedule? Depending on where i put the above, i get errors such as _"NameError: name 'mp_func' is not defined"_ or _"dagster.core.errors.DagsterInvariantViolationError: home_repository not found at module scope in file C:/Users/Jay/PycharmProjects/dagster_home/home_scheduler.py."_ My end goal is to schedule a multithreaded/multiprocessed task (using dask).

Dagster team, can you help me understand why my resources set in the tags section of LaunchPad are not being applied in the run pod? This is resulting in my pods being evicted past the hard limit for ephemeral storage. Config details are below, and you can see the run pod has the limits at lower than what I set. For reference, this is on a GKE managed cluster. Much thanks 🙏🏻

can someone explain how i should use the

snowflake_resource

and

pandas

to construct dataframes, or at least why this doesn't work?

import pandas

from dagster import job, op
from dagster_snowflake import snowflake_resource


@op(required_resource_keys={'snowflake'})
def get_df(context):
    conn = context.resources.snowflake.get_connection()
    return pandas.read_sql_query('select 1 as foo', conn)

@op
def print_df(context, data):
    <http://context.log.info|context.log.info>(data)

@job(resource_defs={"snowflake": snowflake_resource})
def my_snowflake_job():
    df = get_df()
    print_df(df)

second question. is there a way to specify defaults for a

resource

? The

snowflake_resource

shown here always seems default to a certain set of environment variables (snowflake_username, snowflake_password, etc.). How do i override those settings?

@job(resource_defs={"snowflake": snowflake_resource})
def my_snowflake_job():
    df = get_df()
    print_df(df)

third question, does

resource_defs

overwrite all config variables? If i define default configuration, but supply a

resource_def

, it seems to overwrite my default configs. for example, the below yields the following in dagster

defaults = {
    "ops":{
        "write_df": {
            "config": {
                "table": "my_table",
                "schema": "my_schema",
            }
        }
    },
    "resources": {
        "snowflake": {
            "config": {
                "user": "foo",
                "password": "bar"
            }
        }
    }
}

@job(resource_defs={"snowflake": snowflake_resource}, config=defaults)
def my_snowflake_job():
    df = get_df()
    print_df(df)
    write_df(df)

Hi all! I'm deploying dagit/dagster on docker and I've started getting permission errors when the scheduler starts runs, apparently because my user code image is in a private registry. On the docker host, I'm able to

docker pull

with no problem, so is there some extra config I need to pass to dagit or the daemon so they can access the container registry? (I'm saying "I've started..." because it only appeared since I upgraded to 0.14.3 from 0.13.19 but my config was pretty messy, so it might have been hidden behind other problems)

docker.errors.APIError: 500 Server Error for <http+docker://localhost/v1.41/images/create?tag=49a20c1d&fromImage=registry.gitlab.com%2F[myrepo]%2Fdagster_user_code>: Internal Server Error ("Head "<https://registry.gitlab.com/v2/[myrepo]/dagster_user_code/manifests/49a20c1d>": denied: access forbidden")
  File "/usr/local/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 "/usr/local/lib/python3.9/site-packages/dagster_docker/docker_run_launcher.py", line 149, in launch_run
    self._launch_container_with_command(run, docker_image, command)
  File "/usr/local/lib/python3.9/site-packages/dagster_docker/docker_run_launcher.py", line 107, in _launch_container_with_command
    client.images.pull(docker_image)
  File "/usr/local/lib/python3.9/site-packages/docker/models/images.py", line 444, in pull
    pull_log = self.client.api.pull(
  File "/usr/local/lib/python3.9/site-packages/docker/api/image.py", line 428, in pull
    self._raise_for_status(response)
  File "/usr/local/lib/python3.9/site-packages/docker/api/client.py", line 270, in _raise_for_status
    raise create_api_error_from_http_exception(e)
  File "/usr/local/lib/python3.9/site-packages/docker/errors.py", line 31, in create_api_error_from_http_exception
    raise cls(e, response=response, explanation=explanation)

How to you pass a

pandas.DataFrame

to a

graph()

when testing? The “old”

execute_solid()

works for me:

result = execute_solid(
    merge_fact_entitlements,  # a graph() expecting 2 dataframe inputs
    mode_def=inmemory_mode,
    raise_on_error=False,
    input_values={
        'df_from_booking_orders': df_bo,  # a pandas DataFrame
        'df_from_purchase_orders': df_po,  # a pandas DataFrame
    },
)

But trying to do the same thing using

<http://graph.to|graph.to>_job().execute_in_process()

fails whilst trying to coerce the Dataframe inputs (see 🧵 for full stacktrace)

result = merge_fact_entitlements.to_job(  # a graph() expecting 2 dataframe inputs
    name="merge_fact_entitlements_job",
    resource_defs={
        'io_manager': mem_io_manager,
        ..snip..
    },
).execute_in_process(
    raise_on_error=False,
    run_config={'inputs': {
        'df_from_booking_orders': df_bo,  # a pandas DataFrame
        'df_from_purchase_orders': df_po, # a pandas DataFrame
    }}
)

Hi all, I'm using the

python_logs

option in the

dagster.yaml

to stream the python logs to

dagit

and it works great. Really pleased about that. I've mounted the

dagster.yaml

in my

dagit

container and doing so, I'm setting a global logging level for all UCD containers. However, I'm wondering if it would be possible to set the logging level for each UCD separately? E.g. by mounting a separate

.yaml

file in each UCD container. Is this at all supported/possible? Thanks, J.

Is there a way to config an op, within job containing lets say 5 ops, to not run in parallel while it executes. For ex., I have 5 ops in a job. These ops are not really dependent on each other to execute and therefore can be executed in parallel. But there is one specific op which has a high memory consumption and therefore, I'd like for that op to be executed solely (not in parallel to any other op) when it's execution begins. Before/After this op is run, other ops can continue with the parallel execution.

Hey 👋 Question regarding the new software-defined assets. Is it possible to define the

namespace

dynamically at execution time? For example, I want to define assets that represent Snowflake tables with the following namespace:

@asset(
    namespace=["snowflake", <database>, <schema>],
)
def users() -> SnowflakeTable:
    ...

In this example, I would like

<database>

and

<schema>

to be set dynamically depending on the execution context. For instance, if this is run in dev or prod, the database and schema would be different. With the current

@op

framework, I’m achieving that by yielding an

AssetMaterialization

which dynamically sets the asset key based on the current context — can we do something similar with the new

@asset

? Or am I thinking about this the wrong way?

How can I define a RetryPolicy for all ops of a job? I have tens of ops and it would be very heavy to have to insert it manually in all of them

set the channel topic: 🙋 Ask your question after checking out our documentation! https://docs.dagster.io/

Hi all... I just started getting this error when I try to run dagit... Any thoughts on the source?

ModuleNotFoundError: No module named 'ipython_genutils'

Is there a recommended way for local dev to spin up dagit & dagster-daemon?

can someone help me switch the email associated with my slack account from my old work email to my personal email?

Hi, I was wondering if there is a way to have config for a job is a mix between

config_from_files

, and

partitioned_config

to illustrate

config=[
    partitioned_config,
    config_from_files(
        [file_relative_path(__file__, os.path.join("..", "..", "run_config", "pipeline.yaml"))]
    )
]

in pipeline.yaml, I want to config resource

io_manager

e.g:

s3_bucket

…

Hello, is there a way to turn off dagster storing the output of the solids in $DAGSTER_HOME/storage? I’m guessing we can configure this section in the dagster.yaml config?

local_artifact_storage:
  module: dagster.core.storage.root
  class: LocalArtifactStorage
  config:
    base_dir: [$DAGSTER_HOME/storage]

or alternatively, is there a way to store artifacts in cloud storage like in an S3 bucket?

Hi, if I want to store the console log messages generated from the python scripts in each

op

in order to review them for a failed run, what is the commonly used method? (either through Dagster or outside)