Hi team, maybe a basic question, it's possible to provide logs without context in an

op

? like

<http://logging.info|logging.info>

instead of

<http://context.log.info|context.log.info>

?

Hi team I have a problem with the serverless deployment. I continue to get a timeout error ( Repository Location Error). Can you help me debug? Thanks for your help!

Exception: Timed out after waiting 315s for server examplelocation-prod-7ae6cb.serverless-agents-namespace-3:4000. Most recent connection error: dagster._core.errors.DagsterUserCodeUnreachableError: Could not reach user code server Stack Trace: File "/dagster-cloud/dagster_cloud/workspace/user_code_launcher/user_code_launcher.py", line 1412, in _wait_for_server_process client.ping("") File "/dagster/dagster/_grpc/client.py", line 170, in ping res = self._query("Ping", api_pb2.PingRequest, echo=echo) File "/dagster/dagster/_grpc/client.py", line 141, in _query raise DagsterUserCodeUnreachableError("Could not reach user code server") from e The above exception was caused by the following exception: grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNKNOWN details = "Exception calling application: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNAVAILABLE details = "failed to connect to all addresses" debug_error_string = "{"created":"@1669296569.032962470","description":"Failed to pick subchannel","file":"src/core/ext/filters/client_channel/client_channel.cc","file_line":3260,"referenced_errors":[{"created":"@1669296569.032961531","description":"failed to connect to all addresses","file":"src/core/lib/transport/error_utils.cc","file_line":167,"grpc_status":14}]}" >" debug_error_string = "{"created":"@1669296569.033740107","description":"Error received from peer ipv4:10.0.29.197:4000","file":"src/core/lib/surface/call.cc","file_line":966,"grpc_message":"Exception calling application: <_InactiveRpcError of RPC that terminated with:\n\tstatus = StatusCode.UNAVAILABLE\n\tdetails = "failed to connect to all addresses"\n\tdebug_error_string = "{"created":"@1669296569.032962470","description":"Failed to pick subchannel","file":"src/core/ext/filters/client_channel/client_channel.cc","file_line":3260,"referenced_errors":[{"created":"@1669296569.032961531","description":"failed to connect to all addresses","file":"src/core/lib/transport/error_utils.cc","file_line":167,"grpc_status":14}]}"\n>","grpc_status":2}" > Stack Trace: File "/dagster/dagster/_grpc/client.py", line 139, in _query return self._get_response(method, request=request_type(**kwargs), timeout=timeout) File "/dagster/dagster/_grpc/client.py", line 129, in _get_response return getattr(stub, method)(request, metadata=self._metadata, timeout=timeout) File "/usr/local/lib/python3.8/site-packages/grpc/_channel.py", line 946, in __call__ return _end_unary_response_blocking(state, call, False, None) File "/usr/local/lib/python3.8/site-packages/grpc/_channel.py", line 849, in _end_unary_response_blocking raise _InactiveRpcError(state)
  File "/dagster-cloud/dagster_cloud/workspace/user_code_launcher/user_code_launcher.py", line 1047, in _reconcile
    self._wait_for_new_server_ready(
  File "/dagster-cloud/dagster_cloud/workspace/ecs/launcher.py", line 289, in _wait_for_new_server_ready
    self._wait_for_dagster_server_process(
  File "/dagster-cloud/dagster_cloud/workspace/user_code_launcher/user_code_launcher.py", line 1396, in _wait_for_dagster_server_process
    self._wait_for_server_process(client, timeout, additional_check)
  File "/dagster-cloud/dagster_cloud/workspace/user_code_launcher/user_code_launcher.py", line 1418, in _wait_for_server_process
    raise Exception(

Hi all - does anyone know the answer to this stack overflow question: https://stackoverflow.com/questions/74561519/how-do-i-load-a-software-defined-asset-in-dagster-from-outside-dagster In short; how do I load a dagster asset from within a jupyter notebook. Idealy I'd have something like:

from dagster._core import Project 
project = Project(...)
df_users = project.load_asset('df_users', materialize=False)

Hi, a quick question: how do you specify where to materialize dagster assets? Currently everything is in some version of

.../my-dagster-project/tmpav872908/storage/{assetkey}

Hi all, We want to load the assets from 2 dbt projects (

VDBT

and

VDBT_STG

) into our dagster repo (v

1.0.16

), however we’re getting the following error:

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

The error is pretty clear, though we haven’t been able to figure out how to solve it. We’ve tried

with build_resources(..) as resources:

but have not been successful with that yet either, and aren’t sure if that’s actually the way to go. Does anyone here know the answer? Minimal non-working example:

from dagster import repository, with_resources
from dagster_dbt import load_assets_from_dbt_project, dbt_cli_resource
import os

project_folder = os.path.join('python', 'dbt')
VDBT_DIR = os.path.join(project_folder, 'dagster', 'dbt', 'vdbt')
VDBT_PROFILES_DIR = os.path.join(VDBT_DIR, '.profiles')
VDBT_STG_DIR = os.path.join(project_folder, 'dagster', 'dbt', 'vdbt_stg')
VDBT_STG_PROFILES_DIR = os.path.join(VDBT_STG_DIR, '.profiles')


@repository
def dbt():
    return [

        *with_resources(
            load_assets_from_dbt_project(
                project_dir=VDBT_DIR,
                profiles_dir=VDBT_PROFILES_DIR,
                key_prefix='vdbt',
            ),
            resource_defs={
                "dbt": dbt_cli_resource.configured(
                    {"project_dir": VDBT_DIR, 'profiles_dir': VDBT_PROFILES_DIR},
                )
            },
        ),

        *with_resources(
            load_assets_from_dbt_project(
                project_dir=VDBT_STG_DIR,
                profiles_dir=VDBT_STG_PROFILES_DIR,
                key_prefix='vdbt_stg',
            ),
            resource_defs={
                "dbt": dbt_cli_resource.configured(
                    {"project_dir": VDBT_STG_DIR, 'profiles_dir': VDBT_STG_PROFILES_DIR},
                )
            },
        ),

    ]

hi #dagster-support can different service accounts be configured for different deployments in kubernetes helm charts? reviewing the schema in “https://artifachub.io/packages/helm/dagster/dagster-user-deployments?modal=values-schema” I see that it can’t. is there any way to do it? thanks

This message was deleted.

Cross posting from our #announcement channel. Apologies for slow response times today and tomorrow, thanks for your patience https://dagster.slack.com/archives/CCCR6P2UR/p1669147456963219f

Periodic Reporting on Jobs, Schedules and Runs I'd like to create a job which will run periodically to query the dagster instance for information like, • all jobs in all workspaces • all schedules associated with those jobs • all run within some recent time range I already have a job which calls

context.instance.get_runs()

and then runs some checks on those jobs, so that should cover the last point above. But it's not clear from the [documentation](https://docs.dagster.io/_apidocs/internals#instance) how I could programmatically get information about workspaces, jobs and schedules from the dagster instance. I can get this information from the GraphQL API (https://docs.dagster.io/concepts/dagit/graphql#get-a-list-of-repositories), but I'd rather have a job collect this info and then "call home". Any advice on this would be very useful.

Trying to set the local_artifact_storage.config.base_dir and comput_logs.config.base_dir parameters at run time. Something like:

run_config = {'local_artifact_storage':
                  {'config':
                       {'base_dir': r'C:\Users\peter\dagster'
                        }
                   }
              }

How can I add this to an @repository ?

@repository
def my_repo():
    return [test_asset1, test_asset2]

Anyone from Elementl’s team. I was going through the October crash course linked here: https://dagster.io/blog/dagster-crash-course-oct-2022 In the section

Visualize the GitHub stars,

there is an indentation error in the code. The last three lines should be indented. Also the line where we defile the

markdown

variable for some reason there are three spaces instead of four

hey all 🙂 I amrying to write an io manager that uses awswrangler/athena under the hood. I need to be able to pass in the partition column at `handle_output`/`load_input` time as it will be unique to the op. I am thinking that if I used a multipartition with a single partition def I can give the dim a name and access that via the io_manager. not sure if that will work and seems a bit hacky. is there a better way without multiparition?

Hey 👋 I'm trying to link two graphs together so that one is executed after the other. I'm stuck trying to pass a Nothing input to the second graph. I'm getting the following error:

dagster._core.errors.DagsterInvalidDefinitionError: @graph 'graph_two' has unmapped input '_in'. Remove it or pass it to the appropriate op/graph invocation.

Any suggestions? 🙂

@graph
def graph_one():
    pass

@graph(ins={"_in": In(Nothing)})
def graph_two(_in):
    _in

@graph
def graph_three():
    graph_two(_in=graph_one())

Hello I'm trying to make my scheduler order partitions from a StaticPartitionsDefinition class on the daily basis, but I end up with just one partition scheduled. Here's my code: partition creation:

partitions_def= StaticPartitionsDefinition(key_list)

partitioned scheduler:

def get_schedule_def(partitions_def, cron_schedule, job, execution_timezone):
    @schedule(name=f'{job.name}_scheduler',
        cron_schedule=cron_schedule,
        job=job,
        default_status=DefaultScheduleStatus.STOPPED,
        execution_timezone=execution_timezone)
    def schedule_def():
        partition_keys = partitions_def.get_partition_keys()
        if len(partition_keys) == 0:
            return SkipReason("The job's PartitionsDefinition has no partitions")
        for key in partition_keys:
            yield job.run_request_for_partition(partition_key=key, run_key=key)
    return schedule_def

partitioned job creation:

job = asset.to_job(name=f"Job_{job_name}", resource_defs=resource_defs,
                    config=config, partitions_def=partitions_def, executor_def=in_process_executor,)

and job variable + get_schedule_def goes to the repository function

We were on 1.0.16 and upgraded to 1.1.3/0.17.3 and our Snowflake jobs seem to be consuming all of the server memory then crashing. Has anyone else seen the same? I don't believe it's a dagster issue. I am also getting it locally, where I have 128gb of ram 😬 . Will have a dig and try and figure out what is going on

Advice on running the same job in staging and production on different schedules in separate deployments? I currently have our Kubernetes-executed job defined in one repo, and added as a code location in both our staging and production Dagster Cloud deployments. I'd like to add a schedule so that prod runs nightly and staging runs each morning. Ideally each deployment will only show one job and one schedule, and I won't need too much code duplication / need to split into too many files. We have an

ENVIRONMENT

env var, so for now I just have a switch based on that when I'm building the schedules, so when the code's loaded by the agent for

ENVIRONMENT=production

we'll get one schedule definition, and another for

ENVIRONMENT=staging

. Is there a better way?

Hi all; keen to learn what types of assets can be previewed in the Dagster UI. I found a way to display dataframes using markdown, but it'd be brilliant to be able to save Plotly charts / arbitrary HTML / Matplotlib figures - is this supported? Consider this usecase

@asset 
def model(X_train:pd.DataFrame, y_train:pd.Series)->LogisticRegression:
    ...
    return my_model

@asset
def confusion_matrix()->plt.Figure:
    ...
    return fig

Hi team. Is it possible to catch the error/exception in a run_failure_sensor to be notified only on certain errors?

Hi all; what's the "right way" to fetch a list of dynamically materialized / observed assets? Looking at the sensor example in the documentation on the dagster website, it seems that event logs/records are used with an EventRecordsFilter in that example? I was wondering if there were other ways? Or if that is really the best practice? I also wanted to ask if that's the case, is there a way to pass in a filter on metadata to the event records filter in some way (e.g. if I store some metadata field for "foo" in an asset materialization, is there a way to use EventRecordsFilter out of the box to filter on only asset materializations that have non null "foo" metadata field?)?

Has anyone deployed Dagster to AWS ECS and had success getting runs to start “quickly” at scale (say, within 10-20sec after the run is requested)? Currently, my team is hosting a gRPC server

dagster api grpc

in a single ECS task we call

user_code

. We’re using the

EcsRunLauncher

to launch jobs, so when a run is triggered through Dagit, a sensor, etc., we provision a new ECS task pointing to an ECR image holding our own code repository in order to do the work. This scales quite well in terms of throughput and resource usage—runs don’t eat each other’s resources since we spin up as many ECS tasks as we like! But it takes about 1-2 minutes for ECS to provision and launch the new task for a run, which is slower than we’d want. On the opposite extreme, I could imagine simply using the

DefaultRunLauncher

in the gRPC task so that runs could start near-instantaneously after being triggered. But a single task, even a large one, probably couldn’t handle the volume of runs we’re working with. Has anyone worked with any sort of in-between? I’m imagining something like an autoscaling ECS service with a bunch of user-code tasks that launch jobs instantly on demand, but how to put this together isn’t obvious to me. A custom run launcher? Or maybe add more

user_code

gRPC tasks and somehow load-balance traffic to those tasks from

dagit

/

dagster-daemon

? Any thoughts welcome 😃daggy 3d

the group zoom function looks great!

Minor annoyance re: local vs deployed Why, locally, can I use the module import using

from .assets import lead_assets

, but in my user deployment I have to use the full path (i.e.

from foo.sales.assets import lead_assets

This is the error I get

ImportError: attempted relative import with no known parent package

I see most examples using the dot notation, but I’m struggling in making this work while deployed (and I’d like to keep local vs deployed consistant) Dagster is deployed to K8s via Helm Chart. Much thanks!

Hey! We have some complex dynamic repositories that might take some time to load properly. Sometimes Dagit will mark them with a

Failed

status and return

DagsterUserCodeUnreachableError

as expected due to timeout. A manual reload will fix the issue. We also increased the timeout itself. Is there a way to auto-reload the failed repositories? Thanks!

Hi all! First of all really enjoying Dagster, ty for the prod! I h ave a question about how to call dinamically generated OPS I have an op that is generating ops according to an input list

@op(out=DynamicOut())
def build_ops(reports):
    for report in reports:
        yield DynamicOutput(report_op_factory(report), report[TABLE_KEY])

----

def report_op_factory(report):
    @op(
        name=report[TABLE_KEY],
        required_resource_keys=DYNAMIC_OP_REQUIRED_RESOURCES,
        out=Out(io_manager_key=BQ_IO_MANAGER_KEY),
    )
    def inner_op(context: OpExecutionContext):
        """Op to download a report from smartly given a URL"""
        smartly: SmartlyReports = context.resources.smartly_reports
        return smartly.download_report(report[URL_KEY])

    return inner_op

Then in order to run all of them at the same time

run(ops.collect())

----

@op(required_resource_keys=DYNAMIC_OP_REQUIRED_RESOURCES)
def run(context, ops):
    for o in ops:
        o()

The issue i have is that the op call o() requires context but i'm not sure how to pass it. I've tried using build_op_context

o(
    build_op_context(
        {
            BQ_IO_MANAGER_KEY: getattr(context.resources, BQ_IO_MANAGER_KEY),
            SMARTLY_REPORTS_RESOURCE: getattr(context.resources, SMARTLY_REPORTS_RESOURCE),
        }
    )
)

But it does not really attach the IO Manager I ask for even when i specified it in the factory, Anybody has faced the same issue?

The dagit logs don't seem to be well parsed by datadog. is there a way to make the dagster server logs (not within the app) to work better with DD? Not the job execution logging.

dagster devil Dagster docs for this link is showing a 404 - https://docs.dagster.io/concepts/partitions-schedules-sensors/schedules

Hi, I’m working with partitioned assets and Spark. I was wondering if anybody had a good solution (with an

IOManager

or something) to determine which columns to partition the persisted delta table based on the Dagster partition. The problem I’m having is that with is that I can’t find a way to encode the column name into the Dagster partition definition without creating my own implementation; I believe partition definitions (except for multi-dimension partitions) are anonymous so for example I would like to have a partition with

{date:2022-11-29}

instead of just

2022-11-29

.

Hi all, how stable is

MultiPartitionsDefinition

? It is flagged as "Experimental"

I have a question regarding DynamicOutput processes: Can I use the

map

method with more than one dynamic output? That is, I'd like to

map

the summation of the squares and cubes below.

@op(out=DynamicOut(int))
def one_through_five():
    for number in [1, 2, 3, 4, 5]:
        yield DynamicOutput(number, mapping_key=f"number_{number}")

@op()
def square_the_number(number: int):
    return number * number

@op()
def cube_the_number(number: int):
    return number * number * number

@op()
def sum_squares_and_cubes(context, square: int, cube: int):
    <http://context.log.info|context.log.info>(f"square: {square}, cube: {cube}")
    return square + cube

@graph()
def process_sum_squares_and_cubes():
    numbers = one_through_five()
    squares = numbers.map(square_the_number)
    cubes = numbers.map(cube_the_number)
    # I want to map this
    total = sum_squares_and_cubes(squares, cubes)

👋 Hey Folks What’s the best way to query lineage information? Ideally I’d like to query for all downstream assets/dependencies from a given asset during the CICD process. This doesn’t seem to be exposable via AssetDefinition currently, nor the built in graphql python library.