Hi team! I am wondering if there is a way to rerun a subset of ops within a graph but with different inputs please?

Hey team! Does the Software Defined Assets API support cross-repo dependencies? • i.e. Could a job B defined in repo B have assets that are downstream of assets defined in job A in repo A? ◦ In this example, we are using OSS Dagster following the Helm deployment referenced in these Docs. ◦ What if repos A and B live under different User Code Deployments? • Are you able to reference an asset in another repo by the asset key? • Within Dagit, can you visualize cross-repo dependencies in the Global Asset Lineage view? I've seen https://docs.dagster.io/concepts/assets/software-defined-assets#defining-external-asset-dependencies, but wanted to clarify on some of the nuances above

Is there a way to terminate every job in the dagster queue? Accidentally scheduled a job with a timeout error and now have a 1000+ job queue. UI seems to only do 25 at a time.

Hi all! Who can help with dagster-pandera? There is a problem with the operation of DataFrameSchema when passing the schema to the In / Out configuration op. https://github.com/dagster-io/dagster/issues/10957

I always get confused about when should I use

Partition

Assets or

Partition

Jobs. What are the best practices ?

What’s a good way to dependency-inject resources from a Job run? I have a sensor and depending on which condition triggered the RunRequest, I want to use different resources.

What is the equivalent config for

MultiPartitionDefinition

? For example,, we do have

DailyPartitionsDefinition

->

daily_partitioned_config

I didn't find any

multi_partitioned_config

is it something we can achieve with the current multi-partition implementation? I want to create a multi-partitioned job where I can backfill passing

color

and

date

range for example.

How can I denote software defined assets of output data which is potentially materialized inside the notebook (https://docs.dagster.io/integrations/dagstermill/using-notebooks-with-dagster#using-jupyter-notebooks-with-papermill-and-dagster) bypassing any IO managers? I.e. in case a manual

<http://pd.to|pd.to>_sql

is executed in the noteooks?

My colleague has a problem with

AllPartitionMapping

during upstream backfills. He doesn't want to perform a full backfill, instead he wants to work with let's say 1 week of partitions from 2 months ago. Of course, the

AllPartitionMapping

doesn't like the fact that not all upstream partitions are materialized. Meanwhile my colleague wants it to act like "load all partitions that I materialized previously". I think it should be possible to do with a different partition mapping which would query Dagster for partitions that have ever tried to be materialized (both successfully and not) and then get the latest partition key from this list. Any tips on how to implement this? Edit: I realized the downstream asset should probably just be partitioned too

Hi everyone, just a quick basic question, as I am new to dagster: I need a simple data versioning in the sense that every time an asset is materialized a copy is stored to a "history store" for possible later inspection. Is there a best practice in dagster to achieve that? (Of course I can implement it in the op, but there might be a more elegant way)

Good day everyone! I would appreciate an input on a problem I currently have. I'm running dagster 14.14 on k8s and I need to access ops resource from io manager. This is what I try to do:

@io_manager(required_resource_keys={"k8s_pg"})
def my_iomanager(init_context):
    return ioClass(
        resource=init_context.resources.k8s_pg
    )

Where ioClass has init method

self.resource=resource

. The problem I'm having is that the resource inside ioClass is initialized without the env vars. However, the hard coded parameters of the resource are there. I observe this by logging from inside the ioClass

logging.critical(_self_.resource.__dict__)

, resource params with

os.getenv('xxx')

are None. I've also verified that all the required env vars exist inside my user depl container by executing the shell and running

printenv

. The code that I have works if I execute from localhost with

dagster job execute ....

. I've also tried adding _`executor_def=in_process_executor`_ to my job decorator which didn't change anything. Any help would be appreciated!

Hello, I have a requirement where I have to run a data pipeline for a set of keys. I figured out that I have to use

static_partitioned_config

or

dynamic_partitioned_config

, but currently the

_run_config_wrapper

for these decorators only accepts partition_key as input. Is there a way to access

config_mapping

params in this function, so we can add other config values in

run_config_for_partition_fn

in addition to partition_key. If this is not possible, what is the best way to set up the job. Thanks,

Hello, I'd like to work on the equivalent of the

tag_concurrency_limit

available on Runs but for Ops, in order to limit only part of a Run in terms of concurrency (our use case has a DAG with DBT assets (no limit) and Python API calls that would require rate limiting in terms of concurrency) Is there any pointers on how it would be achievable or where to discuss this kind of architecture concerns ?

Hi, I am writing unit tests to test a few Ops and I am getting this exception, but only when I am running two separate tests. If I comment out one test then I don't get this error. The exception is being ignored, so technically my tests pass but still I am assuming that something is wrong. Have you experiences this before?

Exception ignored in: <generator object build_resources at 0x7f5bfcd93df0>

Traceback (most recent call last):

File ".venv/lib/python3.10/site-packages/dagster/_core/execution/build_resources.py", line 114, in build_resources

list(resources_manager.generate_teardown_events())

File ".venv/lib/python3.10/site-packages/dagster/_utils/__init__.py", line 533, in generate_teardown_events

yield from self.generator

File ".venv/lib/python3.10/site-packages/dagster/_core/execution/resources_init.py", line 279, in resource_initialization_event_generator

yield from manager.generate_teardown_events()

File ".venv/lib/python3.10/site-packages/dagster/_utils/__init__.py", line 533, in generate_teardown_events

yield from self.generator

File ".venv/lib/python3.10/site-packages/dagster/_core/execution/resources_init.py", line 351, in single_resource_event_generator

with user_code_error_boundary(DagsterResourceFunctionError, msg_fn, log_manager=context.log):

File "/usr/lib/python3.10/contextlib.py", line 142, in __exit__

next(self.gen)

File ".venv/lib/python3.10/site-packages/dagster/_core/errors.py", line 184, in user_code_error_boundary

with raise_execution_interrupts():

File "/usr/lib/python3.10/contextlib.py", line 142, in __exit__

next(self.gen)

File ".venv/lib/python3.10/site-packages/dagster/_core/errors.py", line 154, in raise_execution_interrupts

with raise_interrupts_as(DagsterExecutionInterruptedError):

File "/usr/lib/python3.10/contextlib.py", line 142, in __exit__

next(self.gen)

File ".venv/lib/python3.10/site-packages/dagster/_utils/interrupts.py", line 90, in raise_interrupts_as

_replace_interrupt_signal(original_signal_handler)

File ".venv/lib/python3.10/site-packages/dagster/_utils/interrupts.py", line 19, in _replace_interrupt_signal

signal.signal(signal.SIGINT, new_signal_handler)

File "/usr/lib/python3.10/signal.py", line 56, in signal

handler = _signal.signal(_enum_to_int(signalnum), _enum_to_int(handler))

TypeError: signal handler must be signal.SIG_IGN, signal.SIG_DFL, or a callable object

For reference, these are my tests: I am using pytest and pytest-mock to write my tests. Test 1:

def test_launch_cluster_cluster_exists(mocker: MockerFixture):

emr_mock = mocker.patch("dagster_aws.emr.emr.EmrJobRunner")

emr_mock.run_job_flow.return_value = "CLUSTER-ID"

emr_mock.cluster_id_from_name.return_value = ""

with mocker.patch('dagster_aws.emr.emr.EmrJobRunner.cluster_id_from_name', side_effect=EmrError()):

context = build_op_context(

resources={

"emr_job_runner": emr_mock

},

config={

"emr_release_label": "emr-6.8.0",

"cluster_name": "TestCluster",

"master_node_instance_type": "m5.xlarge",

"worker_node_instance_type": "m5.xlarge",

"worker_node_instance_count": 2,

"ec2_subnet_id": "subnet-1293819381907",

"worker_node_spot_bid_price": "0.9"

}

)

cluster_id = launch_emr_cluster(context=context)

assert cluster_id == "CLUSTER-ID"

Test 2:

def test_launch_cluster_not_cluster_exists(mocker: MockerFixture):

emr_mock = mocker.patch("dagster_aws.emr.emr.EmrJobRunner")

emr_mock.run_job_flow.return_value = ""

emr_mock.cluster_id_from_name.return_value = "EXISTING-CLUSTER-ID"

context = build_op_context(

resources={

"emr_job_runner": emr_mock

},

config={

"emr_release_label": "emr-6.8.0",

"cluster_name": "TestCluster",

"master_node_instance_type": "m5.xlarge",

"worker_node_instance_type": "m5.xlarge",

"worker_node_instance_count": 2,

"ec2_subnet_id": "subnet-1293819381907",

"worker_node_spot_bid_price": "0.9"

}

)

cluster_id = launch_emr_cluster(context=context)

assert cluster_id == "EXISTING-CLUSTER-ID"

Hello, when using

dagster.materialize

I am getting an error

E           dagster._check.ParameterCheckError: Param "recon_pipeline" is not a ReconstructablePipeline. Got <dagster._core.definitions.pipeline_base.InMemoryPipeline object at 0x7eff75ff40d0> which is type <class 'dagster._core.definitions.pipeline_base.InMemoryPipeline'>.

I believe this is because I am using pyspark. More details in 🧵

Hello, I am trying to deploy a Github project to Dagster cloud. The deploy triggered via Github actions fails with this root error:

ERROR:root:b"pid 1790 -> /home/runner/work/trx_etl/trx_etl/build/.pex/venvs/b9b8c4d1ace8e0b01c93d90648332402b22cf90d/0c9ffbfa1b4080a12413e6319eb7d3da7eff4f40/bin/python -sE /home/runner/work/trx_etl/trx_etl/build/.pex/venvs/b9b8c4d1ace8e0b01c93d90648332402b22cf90d/0c9ffbfa1b4080a12413e6319eb7d3da7eff4f40/pex --disable-pip-version-check --no-python-version-warning --exists-action a --no-input --use-deprecated legacy-resolver --isolated -q --cache-dir /home/runner/work/trx_etl/trx_etl/build/.pex/pip_cache download --dest /home/runner/work/trx_etl/trx_etl/build/.pex/downloads/resolver_download.fkkuxf9x/opt.hostedtoolcache.Python.3.8.15.x64.bin.python3.8 --requirement /home/runner/work/trx_etl/trx_etl/build/deps-requirements-18a180dd18384e5a8abcf4448807cd0618dfdca8.txt --index-url <https://pypi.org/simple> --retries 5 --timeout 15 exited with 1 and STDERR:\nERROR: Double requirement given: dagster==1.0.15 (from -r /home/runner/work/trx_etl/trx_etl/build/deps-requirements-18a180dd18384e5a8abcf4448807cd0618dfdca8.txt (line 28)) (already in dagster (from -r /home/runner/work/trx_etl/trx_etl/build/deps-requirements-18a180dd18384e5a8abcf4448807cd0618dfdca8.txt (line 24)), name='dagster')\n\n"
81

My

requirements.txt

currently lists:

dagster==1.0.15

dagster-graphql==1.0.15

The project runs well locally. Any hints to what might be going on?

Hello support, do you have recommendation to make linters happy with Dagster? One specific problem is

context

missing from function calls since it’s injected by Dagster.

i have a job where every hour I want to call an api and then write the output to a table in snowflake. I do not think i need an asset, i can’t run a backfill - the API is only “value now”. what’s best practices for this kind of job? is there a simple example somewhere?

Trying to follow this doc https://docs.dagster.io/guides/dagster/airbyte-ingestion-as-code (great writeup by the way) however, i’m not sure how to configure the values under

Streams

in the Airbyte UI : • Destination Namespace • Namespace Custom Format • Destination Stream Prefix Tried looking through the docs/code but wasn’t very clear to me.

Hi, I'm looking at the definition of the class

SensorEvaluationContext

, and I'm confused about the following descriptions: • cursor (Optional[str]): The cursor, passed back from the last sensor evaluation via the cursor attribute of SkipReason and RunRequest • last_completion_time (float): DEPRECATED The last time that the sensor was evaluated (UTC). • last_run_key (str): DEPRECATED The run key of the RunRequest most recently created by this sensor. Use the preferred

cursor

attribute instead. I would like to understand a few things: Question 1: If

last_completion_time

is deprecated, then was is the correct way to get this value? Also, I find that _last_completion_time_ & _last_run_key_ methods are very intuitive and it seems odd that they are being deprecated Question 2: Using the SensorEvaluationContext, is there a way to inspect (e.g. fetch the status) of the runs that were triggered. I need to know, for example, if other runs are completed or still running.

Hi team! I had a SourceAsset defined with metadata before

g_pillar_name = SourceAsset(
    key=AssetKey("g_pillar_name"),
    io_manager_key="pillar_name_io",
    metadata={"pillar_name": "governance"})

Then in the io_manager I would reference it

context.metadata.get("pillar_name")

Does this not work? I swear it worked before. Thanks!

Hi Dagster Team, If I give an asset materialization a metadata entry, what's the easiest way to access those values? E.g.

# in some op
context.log_event(
            AssetMaterialization(
                asset_key=AssetKey("foo"),
                description="Persisted result to storage.",
                metadata={
                     "text_metadata": "Text-based metadata for this event",
                }
            )
        )

# in a sensor (or op execution context)
foo_event_records = context.instance.get_event_records(
            EventRecordsFilter(
                event_type=DagsterEventType.ASSET_MATERIALIZATION,
                asset_key=AssetKey(["foo"]),
            ),
            ascending=False,
        )

# get text_metadata from return foo_event_records

When I print out the event records, I can see the stored metadata values for each event record, but I don't want to parse that text, if there's an easy helper function that will get me the metadata values easily?

Hi , How can we introduce authentication to dagster application? Can anyone point to the right resource? The dagster is running AWS ECS. The first guess is to use N*ginx as a reverse proxy but can anyone point to the right resource so it can be correctly configured.* Thanks

Hey Guys, hey @sandy, I didn’t fully grasp the concept of scheduling yet. Is it possible (like back in Airflow) to set a

start_date

and

end_date

for this schedule? Mind, I ain’t looking for working with Assets here, I just need this schedule to start executing in the past (with past ticks) as part of a backfill logic/mechanism.

from dagster import DefaultScheduleStatus
from dagster import job
from dagster import RunRequest
from dagster import schedule
from dagster import ScheduleEvaluationContext

@job
def test_job():
    op_1()
    op_2()

@schedule(
    name='test_pipeline',
    job=test_job,
    cron_schedule='0 3 * * *',
    default_status=DefaultScheduleStatus.RUNNING,
)
def scheduled_daily_pipeline(context: ScheduleEvaluationContext) -> RunRequest:
    execution_date: datetime = context.scheduled_execution_time

    start: datetime = execution_date - timedelta(days=timedelta_days)
    end: datetime = execution_date - timedelta(hours=1)

    config = {
        'config': {
            'start_dt': start.strftime('%Y-%m-%d'),
            'end_dt': end.strftime('%Y-%m-%d'),
        }
    }

    return RunRequest(
        run_key=None,
        run_config={'ops': {op_name: config for op_name in [op_1, op_2]}}
    )

@repository
def repo():
    return [
        scheduled_daily_pipeline
    ]

Hi! I have a quick question regarding https://docs.dagster.io/deployment/run-retries#configuration. Is it sufficient to add

run_retries:
  enabled: true
  max_retries: 3

in the config file? Because I did that and none of my failing jobs are being retried 🤔

Hello all. We have a Dagster Hybrid deployment using the kubernetes executor. Is it possible to reduce the spin up time of each job pod? It's adding a lot of overhead. I found the Celery documentation but this appears to be for self-hosted only. Edit: I found this piece of info talking about "non-isolated runs", maybe this is related? https://docs.dagster.io/master/dagster-cloud/developing-testing/deployment-settings-reference#non-isolated-runs

Is it possible to set the output of

s3_pickle_io_manager

in a different way other than

s3://<bucket>/dagster/storage/<job run id>/<op name>.compute

? I would like to write my files to S3 in the following manner

s3://<bucket>/year/month_day/<op name>.compute

. I guess that the <job run id> is used so that re-running the same Run over and over again will allow any downstream tasks to process the exact same data. This is similar to what I want but year/month_day would be a bit more human readible than a run ID. I am processing my data on EMR and I don't want to couple my Spark code to Dagster. Therefore, a year/month/day partition style would allow Spark to read data independent of Dagster.

Good morning. I have a question after scouring the documentation: is there a better way to have a cleanup operation run at the end of a

@job

other than defining an

In(Nothing)

input for the cleanup

@op

and passing it the output of the previous

@op

?

Hi all, what version of Dagit do we need to see the freshness policy related UI elements? We are on 1.1.3 atm

Pinging this request to allow protobuf v4. I'm stuck on an old version of Pulumi (and therefore an old version of some providers) due to Dagster pinning protobuf 3. It's possible to rebuild proto files so that they can be used with either proto3 or 4, OpenTelemetry recently made this change. Would that resolve Dagster's need for protobuf 3?