Hi again! I am getting this error on dev (local docker agent) for my

slack_on_run_failure

sensor. Was there some more setup I need to do? It works on staging and prod (k8s agent)

AttributeError: 'NoneType' object has no attribute 'external_repository_origin'
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/errors.py", line 184, in user_code_error_boundary
    yield
  File "/usr/local/lib/python3.9/site-packages/dagster/_grpc/impl.py", line 320, in get_external_sensor_execution
    return sensor_def.evaluate_tick(sensor_context)
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/definitions/sensor_definition.py", line 1268, in evaluate_tick
    result = list(self._evaluation_fn(context))
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/definitions/sensor_definition.py", line 1421, in _wrapped_fn
    for item in result:
  File "/usr/local/lib/python3.9/site-packages/dagster/_core/definitions/run_status_sensor_definition.py", line 485, in _wrapped_fn
    location_name=pipeline_run.external_pipeline_origin.external_repository_origin.repository_location_origin.location_name,

Hello, how do I pass

manifest.json

, an asset, to

load_assets_from_dbt_manifest

? I want to do something like this but it is not working. Am I missing something?

@asset
def dbt_assets(context):
    .........
    return manifest.json


dbt_all_assets = with_resources(
    load_assets_from_dbt_manifest(dbt_assets),
    {'dbt': dbt_cli_resource}
)

Hi, I'm facing an old problem here, but this time I need help to identify the problem. Here's the deal, when an op that downloads a file and loads it on memory using pandas DataFrame executes, I'm getting this error: This used to be RAM memory related, but I got 4GB to use, and when this op runs the deploy it's not even close to 2GB, and the CPU usage is neither, its everything OK, it should be okay running this step, but this error is showing up. Is there any reason this can happen? The other print is how I'm reading the downloaded file.

Hey dagster yay Pretty new to Dagster and scratching my head after trying, reading the docs, and searching through threads here 🙂 How do I create a "placeholder" asset materialization job with some selection of assets that I then can override when calling it from a schedule or a sensor? Or do I need to create jobs for all combinations of asset selections that I might want to run/materialize? An example - I would expect the scheduled run of run_stuff_placeholder to materialize my dbt model "snowflake/mart_xy/f_2" but it still just runs "snowflake/mart_xy/f_1"

run_stuff_placeholder = define_asset_job("run_stuff_placeholder", selection="snowflake/mart_xy/f_1")

@schedule(job=run_stuff_placeholder, cron_schedule="*/5 * * * *")
def configurable_job_schedule():
    return RunRequest(
        asset_selection=[AssetKey("snowflake/mart_xy/f_2")]
    )

Hi. I just installed Dagster in my virtualenv on Windows. I ran the

dagit -f cereal.py

command for the cereal.py file I created and saved but got this error

Using temporary directory C:\Users\josh\Desktop\dagster\tmphvlnxh7o for storage. This will be removed when dagit exits.

To persist information across sessions, set the environment variable DAGSTER_HOME to a directory to use.

0it [00:00, ?it/s]

0it [00:00, ?it/s]

C:\Users\josh\.virtualenvs\venv\lib\site-packages\dagster\_core\execution\compute_logs.py:42: UserWarning: WARNING: Compute log capture is disabled for the current environment. Set the environment variable PYTHONLEGACYWINDOWSSTDIO to enable.

warnings.warn(WIN_PY36_COMPUTE_LOG_DISABLED_MSG)

2022-11-03 12:31:57 +0100 - dagit - INFO - Serving dagit on <http://127.0.0.1:3000> in process 18796

ERROR:    [Errno 10048] error while attempting to bind on address ('127.0.0.1', 3000): only one usage of each socket address (protocol/network address/port) is normally permitted

Hi. In some cases I'd like to add a "manual approval" step to an op. The rationale behind it is that at the end of a run, the results would be published to external entities, and we want to manually check for correctness before. I've found this message from a while ago, which seems to address the same issue. Are there any news on this topic ? Any idea on how I could implement this ?

FYI, some markup is rendering funny on the PyPI page (Firefox/Linux):

Hi all, is there a way to get the

input

asset keys from an Ops ExecutionContext? I see a method to get the outputs, but not he inputs I think I can loop through the "ins" and pass them to:

context.asset_key_for_input

, going to give this a whirl

Hi, is there a way to retroactively add tags to an already completed pipeline run?

I have a daemon crashing for no apparent reason, no errors in the log, but suddenly it shows

Shutting down Dagstere code server for package <x> in process <y>

any hints on why this could be happening?

Hello, We've been running Dagster as a service locally, using the default SqliteRunStorage. Because some processes are slow, for example starting numerous backfills or displaying numerous partitions on Dagit, we want to move to PostgresRunStorage. Is there a way to automatically migrate the historical data from Sqlite to Postgres, so we don't lose the history that was displayed on Dagit ?

Another possibly noob question: If have an assetsensor on a partitioned table_a, and table_a has 2 downstream deps (table_b, table_c) and the assetsensor kicks off a partition run for table_a, will table_b/c automatically run as well for the same partition once table_a completes?

Hi All! I am running meltano EL on dagster using kubernets. For long running process after 2 hours of time I am getting this error: " started a new run worker while the run was already in state DagsterRunStatus.STARTED. This most frequently happens when the run worker unexpectedly stops and is restarted by the cluster. Marking the run as failed." Is there any setting in dagster that will prevent restart of kubernets pods? Please help me out.

https://docs.dagster.io/concepts/logging/python-logging when this mentions that the features for python logging (specifically the integration with python's logging module) are experimental - how new is this? What version of dagster's python module supports it?

Hi folks, I've read the docs, so i apologize in advance if I am being dense in not parsing things properly. I'm evaluating dagster open-source for an on-prem usage situation: cluster with 4 machines, with differing gpu capabilities. I want to support data scientists like myself to develop analysis pipelines through three stages. Stage 1: informal code, but using pure python functions for data and model transformations, run on a sampled smaller data set. Stage 2: convert these to SDA's and run as a pipeline. Part 1: run these pipelines on she smaller sampled dataset to establish we didnt make any mistakes Part 2: run the full pipeline. Now, with some combination of dataset (transformations) and models, choose the best performing one. Part 3: Set up an inference pipeline in prod, and perhaps have a re-training scenario

Now, Stage 1 is simple and can be done even in a notebook. In stage 2 part 1 I can add asset decorators, set up my resources, and still materializing to memory, make sure the pipeline works

My confusion starts with Stage2 Part 2. Once I make sure that the pipeline in stage 2 part 1 works, I would run the full pipeline in CI/CD. The idea here is that I want to run all the experiments for a particular dataset-model configuration out off a particular branch, and that I want to do it one one of these machines as a ci/cd process: so think gitea+woodpecker-ci or gitea+jenkins, or github+jenkins or similar. There are 3 confusions: (1) it seems to be i can run it under dagit, in an appropriate container built separately with dagster and dagit built in, but I am not sure. And in this case it seems that the sqlite database is lost inside the container, ephemerally allive until the container finishes, whereas i really want it to be outside. (2) If i want to fire other post training pipelines, such as, say, a model interpretation run, I need to use sensors and have dagster-daemon running outside. So how do i have dagit in my container process talk to this daemon outside? (3) Finally how do I distribute one of these follow on processes to an appropriate machine. Or for that matter, do I need to handle my original ci/cd model training processes running on a particular machine: can dagster do this for me, or ought this be done by the ci/cd "director" designing where the model training runs?

In stage 3 I have a production pipeline, so it makes sense to have both dagit and dagster-daemon running on my cluster-head machine, and for pipelines to be run directly on this machine. But in this scenario there is also the question: what if i want the inference process on another machine? So i am not sure, but perhaps generally the 2 questions here are (a) how does a ci/cd dagit process in a container communicate with a dagster daemon outside (b) and how can i situate the runner processes elsewhere like another machine and still have them communicate with the server and the database.

I am thinking kubernetes solves some of this, but I do not have the option, or the time-resources to invest in k8s.....

Thanks! And once again, I'm sorry if I have missed something in the docs!! Or somewhere else where this scenario has risen: train informally->pipeline with memory materialization->formal training pipeline with parquet materialization -> inference and continual learning (possibly using partitions)

Super enjoyed the dagster-duckdb blog post and video, BTW!!!

Hi there! How do I create a multi-asset from a graph? • I'm creating a graph that loads sets of data files from s3 to snowflake. • I would like to use the same graph to load different data types, for example

['apple', 'banana', 'pear']

• I would like to materialize all data types at the same time, using a

@multi_asset

- for example to create a

fruit_tables

asset • I know how to create a single asset from a graph using

AssetsDefinition.from_graph()

, but how do I make that a multi-asset?

How to type an arbitrary length tuple? Dagster docs say to follow PEP 484 typing and according to those docs

Arbitrary-length homogeneous tuples can be expressed using one type and ellipsis, for example

Tuple[int, ...]

. (The

...

here are part of the syntax, a literal ellipsis.)

when I try to use this on an op I get the error

E   dagster._core.errors.DagsterInvalidDefinitionError: Invalid type: dagster_type must be an instance of DagsterType or a Python type: got Ellipsis.

using dagster 1.0.10

Is there a way to output custom logs to the dagster SQLite db? Currently we output INFO level logs but I don’t see those in the dagster DB, only dagster’s own event logs.

Not sure if this is the right place to ask this but looking for some help grokking how to think about the assets/jobs/graphs/ops. I / our company is looking to adopt dagster for our data pipelining needs! 🎉 Some context: Our company currently run simulations through a CLI call that requires a

config.json

points.csv

and

sweep.py

where

sweep.py

loads in

config.json & points.csv

which is currently a hard constraint as the simulation CLI isnt under our control and the only parameterization possible is required to be contained within the

sweep.py

. Additionally this python is actually

jython

and not

cpython

so we treat the running of this

simulation.exe --run sweep.py

as a subprocess cli call. Currently we embedded some of the tracking hashes around the naming of the files (

config_{CONFIG_HASH}.json

points_{POINTS_HASH}.csv

sweep_{CONFIG_HASH}_{POINTS_HASH}.py

) What we currently do: we run a cli that consumes a directory with an unknown amount of these

sweep.py

files and submits them to

dask

to run and chain the following pipeline-able operations using dask's as_completed functionality. We use some jinja templating to fill in the

sweep.py

as we need and separate the runs into batches of

points.csv & config.json

within a directory (sometimes ~10k+

sweep.py

within directory) that we can call an

experiment

for lack of a better name. When these run, they create some additional files

meta_{CONFIG_HASH}_{POINTS_HASH}.json

&

output_{CONFIG_HASH}_{POINTS_HASH}.csv

and those get combined down stream into a

final_{CONFIG_HASH}_{POINTS_HASH}.csv

(which cant happen within the jython environment due to pandas requirements so is easiest with dumping intermediate files) with the entire directory combined into

experiment.parquet

. This

experiment.parquet

will eventually have other transformations + reports generated from it (likely larger `@graph`s containing this

@graph

and/or

@asset

) but currently focusing on the main path of execution. What I am trying to do: I want to migrate this to using dagster / local s3 (minio) / postgres / (potentially dbt but still require some investigation) and having trouble with the asset abstraction in relation to this. In a previous thread (before I could spend my time fully implementing) I was pointed at potentially utilizing partitions but those look to be more tailored to time windowing rather than runtime s3 buckets/prefixes (looking to implement but not required). It seems like I could at dagit start/import time read the s3 buckets to create the partitions with static_partitioned_config somehow. Additionally, we would like to be able to re-use a lot of this main

run these sweep.pys

graph-ish thing in larger graphs / jobs. Things I think I have the generally right idea: • Since we would like to reuse the operations that consume a directory of `sweep.py`s, it should be a

@graph

-- would want to share some

config_schema

args along with it so require make_values_resource

graphs cant be called within graphs but can be turned into assets that can be used together I guess? my previous discussion about graph in graphs

• to have full control of what is sent to dask / if using

DynamicOutput

, can't use the

dagster-dask

executor (dagster-dask doesnt support `DynamicOutput` ) but instead submit / gather through using the

dagster-dask

resource Things that arnt quite clear: • When can I and can I not pass

context

to an

op/asset/job/graph

? • should a

@job

only contain plain

def some_job(): graph_1(); graph_2(); graph_3();

with no arguments? can/should a

job/graph

have arguments? • should every

.csv

be its own

asset

? Would that mean a partition for every single

.csv

file (potentially in the thousands very quickly)? So far the

asset

idea seems like it should fit here somewhere but its not super clear where yet, but the asset graph visualization is awesome and would love to be able to take advantage of it! dagster spin

what do people do when testing ops that use attributes of the OpExecutionContext that aren't assigned when using the

build_op_context

helper? it seems that you can't pass in a MagicMock instead because Dagster ensures that an OpExecutionContext is being passed to the op

Is it possible to specify io managers for graph backed assets? I’d like to use the default io manager for the ops within the graph but specify a custom IO manager for the assets produced by the graph.

Suggestions for Custom Schedule I have a requirement to schedule a job once a fortnight, but the fortnights needs to aligned with a organisation's specific pay period. I am already determining the start and end dates of the most recent pay period using

tuple(rrule(WEEKLY, interval=2, dtstart=datetime.date(2022,7,4), until=date))[-2:]

where 04/07/2022 is a known historic pay period start date. I'm considering a sensor which runs once a day and checks if "today" is the Tuesday after the most recent period. If so, yield a RunRequest. Any other ideas?

In the resource when i use os.getenv("") , the db connection works well..I tried to move the details to run_config as per document..but they just get the string value instead of reading from the env

@job(resource_defs={"db_session": db_session})
def job_sample_db_test():
    run_sample_db_test()


job_sample_db_test.execute_in_process(
    run_config={
        "resources": {
            "db_session": {
                "config": {
                    "DB_USER": {"env": "DB_USER"},
                    "DB_PASSWORD": {"env": "DB_PASSWORD"},
                }
            }
        }
    }
)

@resource
@contextmanager
def met_db_session(context):
    try:
        user = context.resource_config["DB_USER"]
        password = context.resource_config["DB_PASSWORD"]
        db = os.getenv("DB_DB", "")
        host = context.resource_config["DB_HOST"]
        port = int(os.getenv("DB_PORT", 54332))
        db_connection = create_engine(f'postgresql://{user}:{password}@{host}:{port}/{db}')
        session_maker = sessionmaker(bind=db_connection)
        session = session_maker()
        yield session
    finally:
        db_connection.close()

Whatever is using os.getenv gets the proper values ,but context.resource_config just get the {"env": "DB_USER"} without actually reading the DB_USER from the environment..what i am missing? The secrets are kuberenets secrets and are available by os.getenv I am just looking for a resource which gets me sql alchemy db connection ,so if there is a better way ,i am down for it...

Hi! Is there any way I separate

asset_name

and

table_name

for asset definition with io_manager? I’ve got 2 tables with the same name in different db/schema and defining them in different dagster repos raises error of duplicated asset names.