I am getting following warning while building my assets

ExperimentalWarning: "asset_key" is an experimental argument to 
 function "OutputDefinition.__init__". It may break in future versions, even between dot releases. 
 To mute warnings for experimental functionality, 
 invoke warnings.filterwarnings("ignore", category=dagster.ExperimentalWarning) or use one of the other methods 
 described at <https://docs.python.org/3/library/warnings.html#describing-warning-filters>.
  return klass(

What does it mean and is this likely to break my system in future ?

Has anyone ever tried to test a

dagster_type_loader

? I have one defined using the decorator:

@dagster_type_loader(...config_schema, resources...)
def remote_path_loader(context, config) -> Path:
  ...pull a file and create a Path...

which is tested as such:

def test_remote_path_loader():
  remote_path_loader(???, {"config_1": "value_1"})

But this just results in

TypeError: 'DagsterTypeLoaderFromDecorator' object is not callable

. I’m also not sure how to build a

StepExecutionContext

for this

Hi! I need some help with my run config. I'm trying to pass some resource configs like this

{
  ops: {
    load_config_from_yaml: {
      inputs: {
        score_config_filepath: esg/ybscore/jobs/idealratings_esg_score_config.yaml
        }
      },
    pillar_scores: {
      inputs: {
        pillar_name: governance
      }
    }
  },
  resources：{
    esg_outputter: {
      config: {
        new_score_versions: {
          metric: '11',
          trbc_industrygroup: {
            score: '11.0',
            pillar: '11.0.0'
          }
        },
        scoring_context: trbc_industrygroup
      }
    }
  },

I'm getting an error

Missing , or : between flow map items

. It seems like it doesn't recognize resources as a keyword? But I think it should, and I've input resource configs for other jobs before. Any pointers? Thanks!

Does anyone know if there is a way to change the default timeout for the io manager ? When an asset size is very large and it takes longer than 60 seconds to write out, the op will fail with a timeout error

howdy! If I have already have dagster cloud ECS installation (provisioned with CloudFormation) with a

dev

and

branch

deployment, how would I add another ECS agent to handle

prod

? Thanks!

Hi folks, Has anybody implemented the scalene or memory_profiler in the dagster pipelines?

Hello, I was wondering if there were best practices around running dagit and user code on AWS ECS. So the question I have is if I should separate the container for dagit and user code into 2 services or can should I those in the same task definition? It seems like keeping the in the same task definition would be the way to go but I'm interested if there are some guidelines about that.

Hi all, I have an asset that I am trying to run with a config file on:

@asset(
    partitions_def=daily, 
    io_manager_key="parquet_io_manager", 
    ins={'test': AssetIn(metadata={'allow_missing_partitions': True})}
)
def load_data(test: pandas.DataFrame) -> pandas.DataFrame:
    return test

@asset(
    partitions_def=daily,
    io_manager_key="parquet_io_manager",
    ins={"load_data": AssetIn(metadata={'allow_missing_partitions': True}, partition_mapping=NDaysPartitionMapping(days=1, offset=0))}, 
    # days - determins number of extra days of data loaded (i.e. days=1 for partition 2018-01-03, will return data from 2018-01-03 and 2018-01-02)
    # offset - offsets the data retrieved by date (i.e. offset = 1 for partition 2018-01-03, will return data from 2018-01-02 )
    config_schema={'cellsSeries' : int, 'cellsParallel' : int}
)
def trailing_window(context, load_data: pandas.DataFrame) -> pandas.DataFrame:
    df = load_data
    <http://context.log.info|context.log.info>(f'Days of data loaded: {df.date.unique()}')
    tidied_raw_data = two_day_data_trial.tidy_raw_data(context, database=df)
    soc_ocv_lut = wr.s3.read_parquet("<s3://base-infra-bostonmodeldevelopment-lq00f0lfzytq/dagster/single_event_data/ocv_soc_lut.parquet>")
    return tidied_raw_data




test_invoked = with_resources(
    [test, load_data, trailing_window],
    RESOURCES_PROD, 
)


@repository
def source_repo():
    return [test_invoked, schedule]

The first asset just loads in data from an s3 bucket and returns a DataFrame, This DataFrame is then inputted into the second asset that partitions it and does some operations. This second asset requires two config schemas ('cellsSeries' : int, 'cellsParallel' : int). How do I pass these? This is the source asset I use:

test = SourceAsset(
    key=AssetKey("test"),
    description="Example partitioned data",
    io_manager_key="parquet_io_manager",
    partitions_def=daily,
)

This is the job scheduler I use:

schedule = build_schedule_from_partitioned_job(
    define_asset_job("main", partitions_def=daily)
)

These are the resources I am using:

RESOURCES_PROD = {
    "s3_bucket": ResourceDefinition.hardcoded_resource(
        "some-url"
    ),
    "s3": s3_resource,
    "s3_prefix": ResourceDefinition.hardcoded_resource("dagster"),
    "parquet_io_manager": s3_partitioned_parquet_io_manager,
}

How can I pass a config file into this setup so that ops called inside the "trailing_window" asset can access values using for example context.op_config['cellsParallel']? Thanks!

Hello everyone, I just have nice chat with OpenMetadata team and I strongly believe on their solution for metadata management. They have just integrated dagster to their connectors (understand the DAGs) but are using Airflow to spin new ingestions. The way they do it is pretty simple, it is writing the logic of a DAG dynamically and then have an interface with Airflow to run it. At the end, they are using the same Architecture as dagster so integrating both should not be complex. API oriented, one central servers dagster call daemon. If anyone is interested in join forces to make it happend, can you tell me ? I agree with them for a POC with only local ingestion. After that, there wil be the dpeloyment.

Any reason why the dbt

target

isn’t accepted by

load_assets_from_dbt_project

? I’m currently facing an issue with that since I need to extract some information from the manifest that changes depending on the target (different dbs for dev and prod, need them to be part of the asset keys). Happy to open a PR if there’s no reason it’s not currently supported.

Hi everyone, I have been wanting to use locationstatechange subscription to monitor the repositories, somehow I haven't been able to make it work. Please let me know if anyone has been able to use it successfully.

Hey all, curious about some functionality related to the

@failure_hook

. I had an instance where a job was invoked but resource creation failed and an error was thrown. I'd expect the failure_alert to be triggered there but nothing happened. Is that the expected functionality? Or is there a way to have that be under the umbrella of the job's failure hook?

[Relates to unit testing and

Context

] Hello there, is there an equivalent of

build_init_resource_context

for

StepExecutionContext

. I’m unit testing a custom IO Manager and getting an error when accessing

_context_.has_partition_key

(InputContext). It seems you can pass a

StepExecutionContext

to

build_input_context

but I’m not sure how to instantiate that class.

I am looking for guidance on multi-asset usage. I have the following asset definitions (simplified) which work well:

@asset
def a():
  same_logic("a")

@asset
def b():
  same_logic("b")

@asset
def c():
  same_logic("c")

@asset(
   ins = {"a": AssetIn(), "b": AssetIn(), "c": AssetIn()},
)
def all(a, b, c):
   process(a,b,c)

I'm wondering if there is a more DRY approach to generating the assets? This doesn't seem to be exactly the use case for multi-assets, since each of a,b,c are generated by independent calls. I've considered programmatically yielding assets instead of using the decorator, but I wanted to check if I was missing anything obvious before heading down that path. I am also not sure of the best way to update

all

when new ins are added. Thanks for the advice!

Hey is it possible to add test to

define_asset_job()

where I can not only materialize dbt models but also test them? So currently I have a sensor that got triggered when dbt model get materialized successfully by dagster. I want the sensor to be trigger not but

dbt run

but by

dbt test

Hi all, I want to transfer data (delta lake tables format) from a DW (Databricks) that live in AWS S3 into an OLTP DB. Ideally the data should be streamed or at least batch processed. Is Dagster the right tool for that job? Is there was study which already shows something similar?

Hi everyone! In a federated architecture, I'm trying to use the grpc server to serve info on a "dbt mini-pipeline" to a central dagster - and from there to orchestrate many "dbt mini-pipelines". The mini-pipeline is actually a pod with everything needed (secrets, mounts, init_containers...) + the grpc server. Is there any way for the central orchestrator to tell each grpc server / full stack to execute the dbt run, instead of running it on other executors that will require to duplicate everything?

going through the update process.

gateway_pings_group_prod = AssetGroup(
    [gateway_pings, get_unique_ip_addresses],
    resource_defs=RESOURCES_PRODUCTION
)
gateway_pings_prod = gateway_pings_group_prod.build_job(
    name="splunk_gateway_pings",
    tags={"notify": "slack"}
)
gateway_pings_schedule = build_schedule_from_partitioned_job(
    gateway_pings_prod
)

how do i do this now? I see that AssetGroup is deprecated and I should use

load_assets_from_package_module

but I don;t see how to set

resource_defs

and how should I rewrite the build_job part? also, i don’t have my assets split into modules. do I need to do that?

[gateway_pings, get_unique_ip_addresses]

are just 2 assets in a file with several others.

Hi Ismael. This seems to be a nuance of the SMTP library. I think you need to make some configuration changes to Google to enable sending emails: https://stackoverflow.com/a/33932499/18247749

Cross posting this from #dagster-feedback: https://dagster.slack.com/archives/C01U5LFUZJS/p1664380059966039

can somebody help me in connecting dagster to aws s3 bucket and then upload and fetching a file from it. I am completely new to dagster and this is my first project so please help me

in dagster cloud, we use github actions to build and push a new image. It looks like 1 of the repos (usually a different one each time) just hangs forever. the github action never finishes.

Hi Support, I'd like to use dagster as a tool to automate stuff between online services, e.g. TEAMS, GITHUB, ClickUp, ... I know this is not the original usecase of dagster, but we are users of dagster for ETL, so why using yet another tool. The only thing I see missing to get my use case implemented is the ability to listen to webhooks. I'd like to have some kind of webhook-sensor which NOT works via pulling but constantly listens to new requests. Ideally it would receive some json payload and forward it to the jobs triggered. Anything like that planned or some kind of workaround available? I am aware of the dagit GraphQL interface. But having dedicated webhook sensors would be much more easy/clear and manageable.

hi everyone, I have an asset which depends on two (source) assets, one source for all partitions before 2022-01-01, and the other source for all partitions from 2022-01-01 onward. Is there a support for this scenario? I'm thinking of modifying the io_resource which loads the source asset from different locations depending on values of the partition key, but wonder whether we have a built-in support? Thanks!

Hi all! I'm excited to get us on the latest dagster but having some problems managing my charts. I'm trying to understand the meaning behind this paragraph in the documentation.

User code deployments can be updated independently from other Dagster components, including Dagit. As a result, updates to repositories can occur without causing downtime to any other repository or to Dagit. After updating, if there is an error with any repository, an error is surfaced for that repository within Dagit; all other repositories and Dagit will still operate normally.

According to this thread https://dagster.slack.com/archives/C01U954MEER/p1650362406805329 it basically seems that this is not true? Is there some other meaning behind the statement? Best I can ascertain is that it alludes to the fact that the docker image behind a user code repository could be changed, however the daemon/dagit doesn't pick up changes in that case and that goes against the recommendations in the docs to note use a fixed tag "we recommend using a unique tag (ie not "latest")." Ultimately, I would like do what the docs seem to state. Ideally, I'd like to separate my core dagster deployment (dagit, daemon) from each individual user code repository. To note, I am attempting to use the dagster and dagster-user-deployment helm charts.

Hello, I have quite recently received a requirement to host Dagster as a service over https. now I know that dagster is capable of listening to a port over http, but in order for it for the dagster link to be secure, my organization has recommended me to have dagster hosted over https thereby having secure access and a domain alias instead of the traditional http:/<IP Address>:<PortAddess>. To do this, I have gotten the required ssl certificate keys to open a listener over https but what i do not know is how do i host dagster as a service over the ssl certificate that i have. I searched the dagster docs but was unable to get this information. Therefore, I would like to ask if this is a configuration to be done in dagster or if there is a way to implicitly mention this in the dagit command. For either case, it would be great if any one could either let me know on further steps or point me to the documentation Kindly help in getting this clarified. Thanks and Regards, Ashley Dsouza

Hey all, is there any way to launch jobs async? I'm using threads but it looks like dagster really doesn't like that. The threads continuously receive Interpreter shutdown exceptions, whereas synchronous execution has no issues. Essentially I have several use cases for: have a job that managers other jobs and was hoping to figure out a way to do that without potentially shelling out and submitting jobs. Is this possible? I seem to remember doing this with the legacy pipeline_run but I'm not sure if that's the best path forward.

Howdy! Question for other hybrid users. I'm curious if any others have figured out a good way to setup dynamic values in your

dagster_cloud.yaml

file (or a way to just configure multiple files)? We have different container registries as well as different secret tags we'd like to pass in on a per environment basis. Ex. Our production

dagster_cloud.yaml

would look like:

locations:
  - location_name: workflows
    code_source:
      package_name: workflows
    build:
      directory: ./
      registry: <http://123456789.dkr.ecr.us-east-1.amazonaws.com/production-deployments|123456789.dkr.ecr.us-east-1.amazonaws.com/production-deployments>
    container_context:
      ecs:
        secrets_tags:
          - my_secrets_with_production_tag

and our dev

dagster_cloud.yaml

would look like:

locations:
  - location_name: workflows
    code_source:
      package_name: workflows
    build:
      directory: ./
      registry: <http://123456789.dkr.ecr.us-east-1.amazonaws.com/dev-deployments|123456789.dkr.ecr.us-east-1.amazonaws.com/dev-deployments>
    container_context:
      ecs:
        secrets_tags:
          - my_secrets_with_dev_tag

And so on for additional environments we have....does anyone have recommendations for handling this? I may be off basis and there's probably a better approach to this

Hello, could someone please point out why is a job config neatly pre-filled in one configured job, but not in the other? Probably I am missing something obvious... I am using

ConfigMapping.resolve_from_validated_config(config)

to pre-configure my jobs. While it "works", as in, the configuration with

"geofilter": "bla bla bla"

shows up in the Launchpad in dagit for

job_main_nl_test

, the pre-configuration does not show for

job_main_nl_sample

. Although, the two jobs are defined in the same way, both definitions are in the same module. The difference between them are the assets they contain. I refreshed+restarted dagit several times. job

job_main_nl_sample

does not get it's configuration pre-filled with

"geofilter": "bla bla bla"

in the dagit Launchpad. See the screenshot of the Launchpad.

@config_mapping(config_schema={
    "geofilter": str
})
def config_main_pipeline(val):
    return {
        "ops": {
            'extract_bgt': {
                'config': {'download_dir': '...', 'geofilter': val['geofilter']}},
            'extract_top10nl': {
                'config': {'download_dir': '...', }}},
    }

assets_source_and_input = AssetSelection.groups("source") | AssetSelection.groups("input")

job_main_nl_sample = define_asset_job(
    name="job_main_nl_sample",
    selection=assets_source_and_input,
    config=config_main_pipeline.resolve_from_validated_config({"geofilter": "bla bla bla"})

job

job_main_nl_test

does get it's configuration pre-filled with

"geofilter": "bla bla bla"

in the dagit Launchpad. See the screenshot of the Launchpad.

@asset(
    config_schema={
        "featuretypes": Field(
            list,
            default_value=["gebouw", ],
            description="The feature types to download.",
            is_required=False
        ),
        "geofilter": Field(
            str,
            description="WKT of the polygonal extent",
            is_required=False
        ),
        "download_dir": str
    },
    group_name="group_bla"
)
def bla(context):
    <http://context.log.info|context.log.info>(context.op_config)
@config_mapping(config_schema={
    "geofilter": str
})
def config_test(val):
    return {
        "ops": {'bla': {'config': {'download_dir': '...', 'geofilter': val['geofilter']}}, },
    }
job_main_nl_test = define_asset_job(
    name="job_main_nl_test",
    selection=AssetSelection.groups("group_bla"),
    config=config_test.resolve_from_validated_config({"geofilter": "bla bla bla"})
)

question on dagster_dbt - how can I capture the dbt logs with the dagster provided ops? Ideally I would like the dbt errors as part of the slack error message. I have implemented both run and test jobs as well as a slack sensor but I am not getting the dbt logs. The interface gives me what looks like a dagster error ( in this case one of the tests failed)

dagster._core.definitions.events.Failure: Request aa18527d-91c9-4b5d-8cd1-a1b6aa9d1949 finished with state 'failed' in 82.054154 seconds

  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster/_core/execution/plan/execute_plan.py", line 224, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster/_core/execution/plan/execute_step.py", line 358, in core_dagster_event_sequence_for_step
    _step_output_error_checked_user_event_sequence(step_context, user_event_sequence)
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster/_core/execution/plan/execute_step.py", line 69, in _step_output_error_checked_user_event_sequence
    for user_event in user_event_sequence:
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster/_core/execution/plan/compute.py", line 174, in execute_core_compute
    for step_output in _yield_compute_results(step_context, inputs, compute_fn):
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster/_core/execution/plan/compute.py", line 151, in _yield_compute_results
    user_event_generator,
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster/_utils/__init__.py", line 430, in iterate_with_context
    next_output = next(iterator)
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster/_core/execution/plan/compute_generator.py", line 73, in _coerce_solid_compute_fn_to_iterator
    result = fn(context, **kwargs) if context_arg_provided else fn(**kwargs)
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster_dbt-1.0.5-py3.7.egg/dagster_dbt/ops.py", line 116, in dbt_test_op
    return context.resources.dbt.test()
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster_dbt-1.0.5-py3.7.egg/dagster_dbt/rpc/resources.py", line 340, in test
    return self._get_result(data=json.dumps(data))
  File "/home/ec2-user/dagster/.venv/lib64/python3.7/site-packages/dagster_dbt-1.0.5-py3.7.egg/dagster_dbt/rpc/resources.py", line 561, in _get_result
    f"Request {request_token} finished with state '{current_state}' in "