Is just me or someone else need help to create tests?

Hello Dagster community, I am bit confused about the concept of assets. To me an asset looks like an op. I know dagit treats them differently, but practically as shown in the docs, it seems that assets and ops do the same thing. Please correct me, I am new to dagster. Here an example, a use case and you can suggest where does the concept of assets fits in. I have a system that generates raw data in form of xlsm files (those are excel files with macros). My end goal is to clean, transform the data in these files and load it into a database. here how I go about going this: • I read the excel file (1. function: read_xlsm_file(), it gets a path as input, reads the excel file and returns an excel obj) • Those xlsm file come in form of reports, I need to delete some rows and columns to get only the table (2. function: format_xlsm_file(), it gets an excel obj as input, deletes some rows and columns and returns an excel obj) • After that I export the worksheet as csv file (3. function: write_worksheet_to_csv(), it gets an excel object as input, export it as csv file and returns the path of the csv file ) • Now I have a clean csv file, I need to read as a dataframe and transform it (4. function: read_dataframe(), it gets the path of the file as input, it reads the csv file, tranforms it and returns a dataframe) • Now I have a transformed, ready to be used dataframe, I need to load it to a database (5. function: load_dataframe(), it gets a dataframe as an input and loads it the database) The process goes pretty will, but no Dagster is yet involved. I imagine, I can run all of these functions as ops in a job. But in your opinion, which function should be handled as an asset? Or where does the concept of assets makes sense here in this scenario? And why? Thanks!

Hi Dagster Community, I'm new to Dagster, and have an orientation question. I've modelled my use case as a set of software defined assets. From dagit I can materialise the dag from the global lineage page and everything works quite well. I think the next step would be to define a job based on assets I've created. https://docs.dagster.io/concepts/ops-jobs-graphs/jobs#from-software-defined-assets What I fail to see at the moment is how I can integrate testing and data assertions into the resulting job. After materialisation I'd like to use a tool like great expectations or soda to perform assertions and if these assertions pass then proceed to materialise the downstream dependancies of the asset else fail and alert. I can see how an example of this might work here: https://docs.dagster.io/integrations/great-expectations#using-dagster-with-great-expectations One option would be to construct the job from ops https://docs.dagster.io/concepts/ops-jobs-graphs/jobs#from-software-defined-assets which would allow me to define ops that conducted the testing and put me in more direct control of the Dag. This feels like it takes away from some of the benefits of SDA. Another option would be to use Dagster Types however these feel more designed to check schema at runtime rather than perform assertions. Any pointers from more experienced dagsters would be much appreciated! 😄

In last`weeks community video on youtube, a tighter airbyte integration was announced. Wenn will this be available (that airbyte can be configured via dagster?

Hello Dagster peeps. I'm looking to try out dagster cloud, with agents deployed in a private GKE cluster. I know the docs are more oriented around AWS integrations, but I wonder if anyone has any insights around how to set up an agent in a k8s cluster behind a load balancer that limits scope to just dagster cloud ingress?

Hi, I am getting this error after upgrading from 0.15 to 1.0 branch. I get a message in saying GraphQL error and the following error details. Request help. Operation name: PermissionsQuery Message: Cannot query field "disabledReason" on type "Permission". Path: Locations: [{"line":11,"column":3}]

Is it possible to define custom io_managers for graph-backed assets? If so how can it be achieved?

Is there a way to set the default io_manager (for everything) without having to configure it in every job/asset definition? I'm currently doing this on every job:

Hey yall - do any of you output the actual SQL that is being executed by DBT in your dagster logs? If so, how exactly do I enable that?

Hi folks. I'm trying to figure out repository lazy loading with assets, as our startup time has dramatically slowed down with the introduction of assets (likely mostly due to loading assets from dbt projects, but I haven't profiled this). The docs on this look a bit out of date, but am I correctly interpreting that I need to create a

RepositoryData

class instance for this? What confuses me is that I don't see where the asset definitions are actually populated? https://github.com/dagster-io/dagster/blob/master/python_modules/dagster/dagster/_core/definitions/repository_definition.py#L468. There's also the dictionary style of repository definitions...but those don't look like they support assets either? https://github.com/dagster-io/dagster/blob/1f68be035588dddb4a02c82ad008d1c41dc8241[…]ules/dagster/dagster/_core/definitions/repository_definition.py. The

CachingRepositoryData

class isn't exported, but appears to supports assets. https://github.com/dagster-io/dagster/blob/master/python_modules/dagster/dagster/_core/definitions/repository_definition.py#L482 What's the recommended way to proceed here? Switching to assets from ops brought the startup time for dagster-daemon/dagit up to almost 1 minute from nearly instantaneous, and with the k8s job launcher, that will add almost a minute to every job run.

Hi team! Is it possible for the output of an

@op

be included in the

config_schema

of another

@op

. Backstory, I'm trying to write a sensor that runs a job when the timestamp is greater than the some value. However, the timestamp is the output of a.

@op

function. So, I'm a bit confused because the RunRequest has

run_config

but there wasn't a

config_schema

used. Any thoughts?

I'm having trouble with sensors and `SkipReason`s - I return one, but I still see

Sensor function returned an empty result

rather than the message passed in the skip reason. Am I missing something? This is what is shown in the examples, but just doesn't seem to work in reality. Note that I've mostly been using multi asset sensors, so perhaps this works with regular asset sensors but not multi asset?

Reposting for @Maverick Aaron Lising :

I was having a hard time to open/connect to our dagster localhost:3000. I think we are using to old version of dagster since we started using it since 2020.

On connecting to dagster localhost, if there are pipelines that are unsuccessful/failed, we tend to run it manually on dagster UI via localhost:3000.

Steps on connecting to dagster: (1) open SDK shell terminal; (2) authenticate, (3) paste the ssh link from GCP VM instances

Before, I was able to connect and open dagster localhost:3000, but now, there’s seem a problem and I can’t find a way to solve/debug it.

I noticed in the terminal some warnings (image 2) and don’t even know if that’s related to my problem.

Is/are there a way for me to debug or connect to the dagster UI/localhost:3000? (image 1)

I’ve recently heard of the dasgter cloud, if we had a existing repo, can we connect it to the dagster cloud even if the dagster version was more early than the recent ones? Also, is it possible to get a demo of dagster and dagster cloud somehow?

In a @repository definition, is there a quick way to include all the software-defined assets from the same file rather than maintaining the decorated function's return list myself? Wondering if there might be a collection of Assets I can refer to. I just have a single file right now, currently sketching out my asset graph and trying out options for key prefixes, group names, etc, so maintaining the repo's asset list adds a bit of time. Not a big deal

I'm passing a dict to a sagemaker estimator. I'm receiving an error on the dagster run "Cannot modify ReadOnlyDict". I've even tried making a copy of the received the same error. Any suggestions?

Hi All, Is there a way to reload

dagster.yaml

from

$DAGSTER_HOME

via GraphQL? Looking for something similar to how we can reload a Workspace with a GraphQL request.

Are there any concrete examples of using

dagster_databricks

and the

databricks_pyspark_step_launcher

? I can see one for EMR: https://github.com/dagster-io/dagster/blob/1.0.13/examples/with_pyspark_emr/with_pyspark_emr/repository.py It would be nice to see an example of the config required to pass to:

"pyspark_step_launcher": databricks_pyspark_step_launcher.configured(        {
            # ??? 
        }
    ),

hello, I’ve deployed dagit + dagster via helm chart on k8s, everything works but from postgres (deployed on azure) logs I see a lot of these errors:

2022-10-18 11:00:04 UTC-634e8734.2a168-LOG:  connection authorized: user=dagster_proddatabase=dagster_prod SSL enabled (protocol=TLSv1.2, cipher=ECDHE-RSA-AES256-GCM-SHA384, compression=off)
2022-10-18 11:00:05 UTC-634e8734.2a168-LOG:  could not receive data from client: An existing connection was forcibly closed by the remote host.

any suggestions?

Hello everyone, Is there any way I can skip the next scheduled run if a previously scheduled run isn’t completed yet?

Hi, I trying to make a sensor with lazy loaded but I don't know how to make a sensor. Could you recommend me?

Just starting out. I have a question about moving to persistent storage using postgres. I would like to use postgres for everything (event logs, schedule, run etc). In this documentation: https://docs.dagster.io/deployment/dagster-instance#dagster-storage It lists the dagster.conf config as:

storage:
  postgres:
    postgres_db:
      username: { DAGSTER_PG_USERNAME }
      password: { DAGSTER_PG_PASSWORD }
      hostname: { DAGSTER_PG_HOSTNAME }
      db_name: { DAGSTER_PG_DB }
      port: 5432

But in the dagster-postgres library documentation: https://docs.dagster.io/_apidocs/libraries/dagster-postgres it lists each as separate:

event_log_storage:
  module: dagster_postgres.event_log
  class: PostgresEventLogStorage
  config:
    postgres_db:
      username: { username }
      password: { password }
      hostname: { hostname }
      db_name: { db_name }
      port: { port }

What's the difference? Is

storage

config equivalent to adding the three:

event_log_storage

,

run_storage

and

schedule_storage

?

Question related to environment variables: Which container do they need to be deployed in to be referenced in user dag code? Will the variables being defined in the Helm values at

dagster-user-deployments.deployments[0].env

be sufficient? Specifically, in this Slack example, it’s referencing a SLACK_TOKEN from the environment, but I just want to be 100% sure where these variables should be loaded.

Hello! I’ve created an IO Manager that creates a sqlite database for each dagster run and reads and writes dataframes to the database. The IO Manager works well but when I rerun a portion of the DAG I get this error:

dagster._core.errors.DagsterInvariantViolationError: Attempting to access run_id, but it was not provided when constructing the OutputContext

and this warning:

No previously stored outputs found for source StepOutputHandle(step_key='asset1', output_name='result', mapping_key=None). This is either because you are using an IO Manager that does not depend on run ID, or because all the previous runs have skipped the output in conditional execution.

I only get this error when I rerun assets. Is it possible to create an IO Manager that doesn’t overwrite assets on each run? I’m using

context.get_identifier()

to get the run_id in the IO Manager.

Hello, dagster people I want to know how to generate sensor name dynamically with sensor factory. I want to use each individual sensor to monitor individual dbt test models, hence I need a unique name for each sensor with the current setup like below, dagit tell me that I have duplicated definition

my_sensor

#my sensor factory
def create_vero_event_sensor(
    request_jobs: Optional[List[JobDefinition]],
    run_status: DagsterRunStatus,
    monitored_jobs: Optional[List[JobDefinition]],
    sensor_name: str,
) -> RunStatusSensorDefinition:

    @run_status_sensor(
        monitored_jobs=monitored_jobs,
        run_status=run_status,
        request_jobs=request_jobs,
    )
    def my_sensor (context: RunStatusSensorContext):
        message = f'Job "{context.dagster_run.job_name}" succeeded.'
        print(message)
    return my_sensor

# loops that factory loop through
test_sensors: List[Tuple[UnresolvedAssetJobDefinition, JobDefinition, str]] = [
    (
        my_dbt_test1,
        job1,
        "sensor_name1",
    ),
    (
        my_dbt_test2,
        job2,
        "sensor_name2",
    ),
]

dbt_sensors: List[RunStatusSensorDefinition] = []
for my_test, jobs, sensor_name in test_sensors:
    sensor = create_vero_event_sensor(
        monitored_jobs=[my_test],
        run_status=DagsterRunStatus.SUCCESS,
        request_jobs=[jobs],
    )
    dbt_sensors.append(sensor)

Question, I need to use an SDK for nats.io that requires

asyncio

for everything, but dagster seems to lack support for

asyncio

with `@sensor`s. So basically calling any nats function (like creating a connection) requires an

await

keyword, which can only be used within an

async def

function. But dagster sensor needs to be set up as a synchronous

def

function, which makes it very difficult to integrate with the nats sdk. Any suggestions on how have an

async @sensor

or how to work around?

is there a list of things to avoid when doing multi-threaded work within an op? The context is that we are using dask's ThreadPoolExecutor to spawn 4 threads. We noticed logs were being dropped, which led me to this related thread where I learned the context object not being thread-safe. A related question is, if we want to have threadsafe logging, should we use the default logging module instead of the one provided by context?

Using

op_tags

on an

@asset

to configure

pod_spec_config

for the

K8sRunLauncher

doesn't seem to be working. Is this something that should be possible?

The Dagster Cloud alert emails tag doesn't seem to be functioning the way described in the docs. We have jobs failing that have the

dagster-cloud/alert_emails

tag, but no alert emails are being sent. Docs: https://docs.dagster.io/dagster-cloud/account/setting-up-alerts#using-system-tags-to-configure-alert-emails Any suggestions?

Partial reruns of a job using `execute_in_process I'm building and running a job in a notebook as part of my development workflow. I'm specifying the

io_manager

as

fs_io_manager.configured({"base_dir": "storage"})

and the op outputs are being pickled as expected. I'd like to be able to run only a subset of the job's ops, but using the outputs from a previous job as the inputs, similar to what I can do in

dagit

where I can re-execute selected ops. A justification would be not having to run an expensive API call again and again. ❔ When calling

execute_in_process

, I know I can use

op_selection

to specify the ops I want to run, but how do I reference the job whose op outputs should be used? https://docs.dagster.io/_apidocs/execution#executing-jobs

I’ve got a graph-backed asset that represents an operation that I’m performing in multiple environments. I’ve recently refactored all of the environment-specific stuff into resource configuration instead of op/graph configuration and I’m running into issues where the resulting assets from

AssetDefinition.from_graph

no longer have independent keys and I don’t see an obvious way to set them. e.g. this results in multiple named

{env}_db

assets due to the use of

configured

refresh_assets = [
    AssetsDefinition.from_graph(
        refresh_db.configured(
            {"environment_name": env}, name=f"{env}_db"
        ),
        resource_defs={
            "db_engine": resources.dev_db_engine,
        },
        group_name="dev_env_databases",
    )
    for env in ENVS
]

this results in duplicate asset keys because there’s no way to set a

name

or

key

in

from_graph

refresh_assets = [
    AssetsDefinition.from_graph(
        refresh_db,
        resource_defs={
            "db_engine": resources.dev_db_engine.configured({"environment_name": env}),
        },
        group_name="dev_env_databases",
    )
    for env in ENVS
]

using

with_resources

instead of the experimental

resource_defs

also doesn’t seem to provide a way modify the asset name/key anything I’m missing?