dagster #dagster-support

Mat Brady

10/05/2022, 1:57 PM

Oh and thanks! 🙂

Balázs Dukai

10/05/2022, 2:19 PM

I'm just going to put this here for others if they get some unexpected resource configuration behaviour. I manage my resource configs by the deployment type (as per the fully_featured_project . So there is s resource-selector in my

repository.py

Copy code

resource_defs_by_deployment_name = {
    "prod": RESOURCES_PROD,
    "local": RESOURCES_LOCAL,
}


@repository
def repo():
    deployment_name = os.environ.get("DAGSTER_DEPLOYMENT", "local")
    resource_defs = resource_defs_by_deployment_name[deployment_name]

    definitions = [
        with_resources(all_assets, resource_defs),
        *all_jobs
    ]

    return definitions

Then the

RESOURCES_LOCAL

contains the resource definitions for the local environment, and it is defined together with all the resources in

resources.___init___.py

(this is wrong!)

Copy code

from resources.files import file_store
from resources.database import container

file_store_temp = file_store.configured({})

RESOURCES_LOCAL = {
    "file_store": file_store_temp,
    "container": container,
}

Where

file_store

and

container

are resource definitions. While

file_store_temp

is a configured resource. Now, here is where I stumbled (for many...many...hours). I have a job,

job_samle_data_image_test

, which is run by a

run_status_sensor

. The important bit is the

run_config

that goes into the

job_sample_data_image_test

that is executed by the

RunRequest

Copy code

@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
    name=f"asset_testing_success",
    monitored_jobs=[job_test, ],
    request_job=job_sample_data_image_test
)
def sensor_asset_testing_success(context):
    # The container and the temp dir in the monitored job were created with the
    # Run Id of the monitored job. In order to find the container the dir path, we
    # need the Run Id of the monitored job.
    run_id = bcore.get_run_id(context, short=True)
    container_id = make_container_id(run_id)
    temp_path = make_temp_path(run_id)

    image_tag = f"{datetime.today().date().isoformat()}-{run_id}"

    if context.dagster_run.job_name != job_sample_data_image_test.name:
        run_config = {
            "ops": {
                "sample_data_image_test": {
                    "config": {"image_repository": "test/sample-data",
                               "image_tag": image_tag,
                               "image_data_dir": "/tmp"}}
            },
            "resources": {
                "file_store": {"config": {"data_dir": temp_path}},
                "container": {"config": {"id": container_id}},

            }
        }
        return RunRequest(run_key=None, run_config=run_config)
    else:
        return SkipReason("Don't report status of status_reporting_job")

Notice that in the

run_config

I explicitly configure the two resources. When the

job_sample_data_image_test

is run, guess what resource configuration values is it going to take? My guess was that it will take whatever I pass in the

run_config

, thus

"file_store": {"config": {"data_dir": temp_path}}

and

"container": {"config": {"id": container_id}}

. Well, it turns out that this is just half-true. The job will get a

"file_store": {"config": {}}

, because that is what is configured for
file_store_temp
, which is passed to
RESOURCES_LOCAL
, which apparently overwrites the
run_config
for the job. Then the

container

will get

{"id": container_id}

from the

run_config

, because RESOURCES_LOCAL only has the resource definition, and not the configured resource. That's it, I hope it can save some time for others.

🤖 2

Zachary Bluhm

10/05/2022, 3:04 PM

Hi all, is there any way to force backfills to run partitions sequentially? The use case is that some of our models contain lookback aggregations, and these will not be correct unless the backfill is performed sequentially Or better yet, is there a way to specify that a group of assets depends on [current_partition - 30 days, current_partition] of their upstreams being completed

Félix Tremblay

10/05/2022, 3:10 PM

Hello 👋 I'm looking for recommandations regarding structuring dagster pipelines in monorepo vs. multirepo. There is not many information online, besides this discussion (https://github.com/dagster-io/dagster/discussions/8972)

Alexander Voishchev

10/05/2022, 3:12 PM

Hello everyone. Is there any way to get the current asset data? I need to materialize some asset as DataFrame, not to replace (i.e. delete and insert), but to append to the current dataframe (database table), so each materialization will increase the volume of the data of the asset. For example, I need to append to the dataframe (database table) some data from the external source:

Copy code

@asset
def asset1():
    df = get_external_data()
    return df

@asset
def asset2(asset1):
    # Load the current data of asset2 (state from previous materialization)
    df = get_asset2_data()
    # Append the data from asset1
    df = pd.concat(df, asset1)
    # Return the comdined result
    return df

Is there any elegant way to do that without custom IOManager? P.S. I found an example with an asset which returns itself data, as far as I understand:

examples/docs_snippets/docs_snippets/concepts/assets/asset_w_context.py

Copy code

@asset(required_resource_keys={"api"})
def my_asset(context):
    # fetches contents of an asset
    return context.resources.api.fetch_table("my_asset")

But it is an abstract code, as I wrote above.

Sean Lopp

10/05/2022, 4:33 PM

I'm attempting to use the new

load_asset_value

functionality in a notebook as an aide for local development. I am able to import my repository successfully, and I confirm in the notebook that

DAGSTER_HOME

is set, but when I attempt to load the asset I get a file not found error because it is attempting to read from

/tmp

. Perhaps I am not understanding how assets are loaded outside of a running dagster context, but is there a way to specify the storage path that was used for materializations?

daniel

10/05/2022, 5:30 PM

Hey Saul - that looks to me like your sensor is taking more than 60 seconds to execute a tick and is hitting a timeout. There's a bit of guidance in how to handle this in the docs here: https://docs.dagster.io/master/concepts/partitions-schedules-sensors/sensors#evaluation-interval - we recommend breaking up the work that each tick does into chunks and using a cursor to keep the amount of time that each tick takes from growing unboundedly

James Hale

10/05/2022, 6:27 PM

Hello - is there a pattern for sharing resources between Assets/Ops and IOManagers? E.g., I'd like to share the same connection configuration for Snowflake between an Op (for queries) and the IOManager (to write results to a table) that is handling the output of that Op. Our current solution below (note the implementation of the Snowflake IOManager and resource are bespoke, not from dagster-snowflake):

Copy code

snowflake_config = {...}

# ...

resource_defs = {
    "snowflake_io_manager": snowflake_io_manager.configured(snowflake_config),
    "snowflake_resource": snowflake_resource.configured(snowflake_config),
}

🤖 1

David Jayatillake

10/05/2022, 9:50 PM

has anyone made a devcontainer for using Dagster with codespaces?

Saad Anwar

10/05/2022, 10:30 PM

Hello! We wanted to reach out to our Dagster contact for some questions regarding application security, but we are not getting any response. Is there some way we can get hold of the contact or some other resource from Dagster? Happy to share the contact name or which Company / Org we are from privately if you would like! Thank you!

🤖 1

Ismael Rodrigues

10/06/2022, 3:20 AM

Hi, good night, About the email on the error sensor, I just found this: https://stackoverflow.com/questions/16512592/login-credentials-not-working-with-gmail-smtp It says Google does not allow to send e-mails using smtplib, that's why I'm getting this error message on Dagster (Picture 1) The main problem is, the 'solution' was to allow the less secure apps so the smtplib would work, but google also removed this option from the account settings as the second picture shows: My question is: How can I use this e-mail on sensor error without having so much bureaucracy to configure and make it work?

teodorpaius

10/06/2022, 8:56 AM

Hello im using dagster together with postgres storage. For whatever reason on iddle the cpu is approx 40% used on postgres side (using 2 vcpus on azure). When running a pipeline the ussage goes to 100%. Do you know how i can throttle this usage on dagster side? Or why i can use so much computing power? Im not logging anything and also im managing my input output for ops inside another db

Davi

10/06/2022, 9:22 AM

Hello, Is there a way of chaining graphs ? Thanks !

George Fourmouzis

10/06/2022, 9:48 AM

Hello! I am trying out dagster and have the following use case: Retrieve an arbitrary number of files from an external service (around 800, but not fixed), process each of those individually, then merge them and perform further processing. How would I approach this in dagster? As far as I understand there isn’t a way to define arbitrary outputs of an asset

🤖 1

Timo

10/06/2022, 10:45 AM

Quick question: What happens if I set the "minimum_interval_seconds" of a sensor to 0? I intend to have a very quick reaction to an external event.

🤖 1

geoHeil

10/06/2022, 1:29 PM

But running a backfill for that specific date worked fine

Huy Dao

10/06/2022, 1:56 PM

Hi everyone, I want to add an op that send out email. Is there a native email or slack integration in dagster? I couldn't find anything in docs. My company is running self hosted dagster and not dagster cloud btw

🤖 1

Huy Dao

10/06/2022, 1:57 PM

A follow up question is: what is the best way to trigger a second dag at the end of the first dag

🤖 1

Josh Clark

10/06/2022, 2:00 PM

I have a question about loading assets values outside of dagster runs, specifically how this works with partitioned files. Using the toy example plus additions I have

Copy code

@asset(partitions_def=DailyPartitionsDefinition(start_date="2022-01-01"), io_manager_key="fs_io_manager")
def my_daily_partitioned_asset(context):
    <http://context.log.info|context.log.info>(
        f"Processing asset partition '{context.asset_partition_key_for_output()}'"
    )

    return [1,2,3,4,5]


assets_with_io = with_resources(
    [my_daily_partitioned_asset],
    resource_defs={
        "fs_io_manager": fs_io_manager.configured(
            {"base_dir": "/Users/josh/Documents/axion/axion-pipeline/assets"}
        )
    },
)


@repository
def repo():
    return [assets_with_io]

But when I try to load (after materialising the asset) with

Copy code

asset1_value = repo.load_asset_value(AssetKey(("my_daily_partitioned_asset","2022-10-05")))

or

asset1_value = repo.load_asset_value(AssetKey("my_daily_partitioned_asset/2022-10-05"))

I get

KeyError: AssetKey

➕ 1

🤖 1

10/06/2022, 2:01 PM

Noob question: Configuring dagster for the first time. The goal is to pull data from a redshift database, shove it through an ML pipeline, and save the results dynamo. I have the assets portion up and running, I've written my op, but I'm having a problem accessing the assets within the op. How does one access assets in an op?

🤖 1

Vinnie

10/06/2022, 2:06 PM

Is it possible to generate an

AssetSelection

that matches all assets that include a specific key without passing the fully qualified name? e.g. I have assets

foo / bar

and

foo / baz

, I’d like to pass

AssetSelection.some_method("foo")

and have it select both assets. I tried passing it to the

AssetSelection.keys()

and it didn’t work. Looking at the source code, there doesn’t seem to be a more permissive method. On a sidenote, I don’t think the

AssetSelection

shows up in the docs website, I just found it by chance somewhere

🤖 1

geoHeil

10/06/2022, 2:22 PM

When launching backfills - how can I ensure the backfills for a particular job are NOT performed in parallel?

Saul Burgos

10/06/2022, 3:09 PM

What are your recommendations to reload lazy loading repositories if the information comes from API and let's say this API is not running at the moment? I am following this example: https://docs.dagster.io/_apidocs/repositories#repositories And this is the pseudocode that I am trying to create

Copy code

class LazyRepositories():
    get_all_jobs(self) -> List[JobDefinition]:
        #response = requests.get('<https://api/jobs>')
        #create dynamic jobs
        #return jobs

    def get_all_pipelines(self) -> List[PipelineDefinition]:
        response = requests.get('<https://api/pipelines>')
        #create dynamic pipelines
        #return pipelines
        
    def get_all_sensors(self) -> List[SensorDefinition]:
        response = requests.get('<https://api/sensors>')

    def get_all_schedules(self) -> List[ScheduleDefinition]:
        response = requests.get('<https://api/schedules>')
        #create dynamic schedules
        #return schedules
        

@repository
def core_generic():
    return LazyRepositories()

If the my API for some reason fails, My questions are: - Should I add the logic on "My dagster repository" to retry again? - Does dasgter has an endpoint to reset all the process? - Should I manage the process of retrying outside of the dagster repostory? I mean.. check if the repository was loaded and if not wait and reload the repository? What I am trying to ask is ....what would be the best approach if my API fails... inside dagster or not?

Carter

10/06/2022, 3:42 PM

Hey guys, just wanted to check in on something - is the dagster context.instance.delete_run(run_id) meant to also clear the local artifact storage for the indicated run if it exists? We recently cleared out a substantial number of older runs from our system and found the local artifact storage wasn't impacted

🤖 1

Steven Tran

10/06/2022, 4:06 PM

Hello! So I have a sensor listening to an event. For this event, I have multiple jobs that will be performed in parallel. Using the default run_coordinator, will all of these jobs be executed in 1 pod that has the user_code or will it be executed on multiple user_code pods?

Madeline Wu

10/06/2022, 6:43 PM

Hello! I am defining a sensor that yields a

RunRequest

, which i I want to run with a

dagster-k8s/config

tag. I’m running into a

CheckError

because the value in the

tags

are expected to be a str but the dagster k8s config is a dict. The relevant source code seems to be here. Is the str typing intended for the

RunRequest

or can we loosen the typing requirement?

🤖 1

James Hale

10/06/2022, 6:44 PM

Hello! Is there a way to have Dagster run a job once immediately upon deploy (or re-deploy)? E.g., creating a database for branch-deployments?

🤖 1

Pablo Beltran

10/07/2022, 1:08 AM

Hey there would there be any issues with using a step launcher in conjunction with dagstermill?

peay

10/07/2022, 10:26 AM

Hello, I am hitting a weird issue with asset configuration. Trying https://docs.dagster.io/1.0.6/concepts/assets/software-defined-assets#asset-configuration with configuration where all fields have default values,

context.op_config

None

when I try and materialize an asset without providing any configuration override. However,

context.op_config

is correctly set if I do override at least one field from the materialize launchpad. Is this a known bug? Running Dagster 1.0.6

Archie Kennedy

10/07/2022, 10:30 AM

Hello. Is there a way to have a

Permissive

dict with 1 required field (op config_schema). Example:

Copy code

@op(
    config_schema={
        "payload": Field(
            Permissive(
                {
                    "URI": Field(str, is_required=True),
                    "Webhook": Field(str, is_required=False),
                },
            )
        )
    }
)