dagster #dagster-support

Ivan Tsarev

11/29/2023, 1:23 PM

Hi folks! I'm in process of transition from legacy Resources to Pydantic-resources and wonder is there a way to pass legacy resource resource as a dependency to ConfigurableResource? Documentation has an example of Pydantic-to-Pydantic dependency, but it's completely unclear if I can pass old-style resource there. I understand that mixing two resource styles is not the best idea overall, but basically has no choice since moving all existing code base to Pydantic at once is quite a time-investment for our codebase.

Davi

11/29/2023, 4:12 PM

Hello everyone, Out of the blue, my k8s config in Dagster stopped working and raises me the following error: Exception: Unexpected keys in model class V1Container: {'pod_spec_config'}

File "/usr/local/lib/python3.10/site-packages/dagster/_daemon/run_coordinator/queued_run_coordinator_daemon.py", line 333, in _dequeue_run

instance.run_launcher.launch_run(LaunchRunContext(dagster_run=run, workspace=workspace))

File "/usr/local/lib/python3.10/site-packages/dagster_k8s/launcher.py", line 282, in launch_run

self._launch_k8s_job_with_args(job_name, args, run)

File "/usr/local/lib/python3.10/site-packages/dagster_k8s/launcher.py", line 204, in _launch_k8s_job_with_args

container_context = self.get_container_context_for_run(run)

File "/usr/local/lib/python3.10/site-packages/dagster_k8s/launcher.py", line 199, in get_container_context_for_run

return K8sContainerContext.create_for_run(dagster_run, self, include_run_tags=True)

File "/usr/local/lib/python3.10/site-packages/dagster_k8s/container_context.py", line 192, in create_for_run

user_defined_k8s_config = get_user_defined_k8s_config(dagster_run.tags)

File "/usr/local/lib/python3.10/site-packages/dagster_k8s/job.py", line 203, in get_user_defined_k8s_config

return UserDefinedDagsterK8sConfig(

File "/usr/local/lib/python3.10/site-packages/dagster_k8s/job.py", line 105, in __new__

container_config = k8s_snake_case_dict(kubernetes.client.V1Container, container_config)

File "/usr/local/lib/python3.10/site-packages/dagster_k8s/models.py", line 116, in k8s_snake_case_dict

raise Exception(f"Unexpected keys in model class {model_class.__name__}: {invalid_keys}")

I am using Dagster with poetry with the following versions:

Copy code

[tool.poetry.dependencies]
python = "^3.9"
dagster = "1.4.7"
dagster-webserver = "1.4.7"
dagster-graphql = "1.4.7"
dagster-postgres = "0.20.7"
dagster-k8s = "0.20.7"
dagster-msteams = "0.20.7"
dagster-gcp = "0.20.7"

And check here my tags configuration:

Copy code

def get_k8s_config(cpu_min, cpu_max, memory_min, memory_max):
    return {
        "dagster/max_retries": 3,
        "dagster-k8s/config": {
            "container_config": {
                "resources": {
                    "limits": {"cpu": cpu_max, "memory": memory_max},
                    "requests": {"cpu": cpu_min, "memory": memory_min}
                }
            },
            "pod_spec_config": {
                "nodeSelector": {...},
                "tolerations": [
                    {
                        "effect": "NoSchedule",
                        "key": "workload-type",
                        "operator": "Equal",
                        "value": "batch"
                    }
                ]
            }
        }
    }

@job(
    resource_defs={
        "msteams": msteams_resource.configured(
            {"hook_url": os.getenv('TEAMS_WEBHOOK_URL')}
        )
    },
    tags=get_k8s_config(
        cpu_min="2500m",
        cpu_max="5000m",
        memory_min="30000Mi",
        memory_max="30000Mi"
    )
)
def job__process():
...

nickvazz

11/29/2023, 5:58 PM

is there a way to make sensor submitted jobs NOT appear as ad_hoc materializations

Alex Chisholm

11/29/2023, 7:34 PM

Hey all, is it normal for a run to need at least 1 minute to allocate resources before it can start running? Anyway to make that faster? Would love to use Dagster but not sure if it can be used for real time data orchestration. Thank you!

Alex Chisholm

11/29/2023, 7:40 PM

I presume if we host Dagster on our own server, it's not an issue, but I'd prefer to use Dagster Cloud.

Anton Podviaznikov

11/29/2023, 7:45 PM

I'm trying to run dagster self hosted in k8s. However when I materlize my job I'm getting

No module named 'dagster_postgres'

inside

dagster-run-...-..

pod. Anyone know why is that and where do I need to add this module?

Copy code

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/dagster/_serdes/config_class.py", line 91, in rehydrate
    module = importlib.import_module(self.module_name)
  File "/usr/local/lib/python3.8/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
  File "<frozen importlib._bootstrap>", line 991, in _find_and_load
  File "<frozen importlib._bootstrap>", line 961, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
  File "<frozen importlib._bootstrap>", line 991, in _find_and_load
  File "<frozen importlib._bootstrap>", line 973, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'dagster_postgres'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.8/site-packages/dagster/_cli/__init__.py", line 48, in main
    cli(auto_envvar_prefix=ENV_PREFIX)  # pylint:disable=E1123
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1157, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1078, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1688, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1434, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 783, in invoke
    return __callback(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/dagster/_cli/api.py", line 64, in execute_run_command
    with get_instance_for_cli(instance_ref=args.instance_ref) as instance:
  File "/usr/local/lib/python3.8/contextlib.py", line 113, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.8/site-packages/dagster/_cli/utils.py", line 104, in get_instance_for_cli
    with DagsterInstance.from_ref(instance_ref) as instance:
  File "/usr/local/lib/python3.8/site-packages/dagster/_core/instance/__init__.py", line 600, in from_ref
    unified_storage = instance_ref.storage
  File "/usr/local/lib/python3.8/site-packages/dagster/_core/instance/ref.py", line 496, in storage
    return self.storage_data.rehydrate(as_type=DagsterStorage) if self.storage_data else None
  File "/usr/local/lib/python3.8/site-packages/dagster/_serdes/config_class.py", line 120, in rehydrate
    return klass.from_config_value(self, check.not_none(result.value))
  File "/usr/local/lib/python3.8/site-packages/dagster/_core/storage/legacy_storage.py", line 122, in from_config_value
    run_storage = ConfigurableClassData(
  File "/usr/local/lib/python3.8/site-packages/dagster/_serdes/config_class.py", line 93, in rehydrate
    check.failed(
  File "/usr/local/lib/python3.8/site-packages/dagster/_check/__init__.py", line 1590, in failed
    raise CheckError(f"Failure condition: {desc}")
dagster._check.CheckError: Failure condition: Couldn't import module dagster_postgres.run_storage when attempting to load the configurable class dagster_postgres.run_storage.PostgresRunStorage

nickvazz

11/29/2023, 8:12 PM

Is it possible to use partitions for observable source assets? https://docs.dagster.io/concepts/assets/asset-observations#observable-source-assets

Drew Broadley

11/30/2023, 12:49 AM

Hi there, I have an asset of "zendesk_reports" which is essentially asset for tickets from zendesk. This asset needs to be shared across multiple pipelines. What's the best approach? The asset is partitioned by days. Defining the asset in a seperate assets.py and importing to each pipeline then sharing means Materialising wont work due to this error:

Copy code

Multiple asset definitions found
One or more of the selected assets are defined in multiple code locations. Rename these assets to avoid collisions and then try again.
zendesk_reports
zendesk_reports_comments_fields_taxonomy_retention@pipelines.i_zendesk_r_reports_comments_fields_o_sql_f_hourly
aggregated@pipelines.i_various_r_aggregated_o_sql_f_hourly

Anant Mishra

11/30/2023, 1:17 AM

Hi everyone, I'm currently working on a project where I need to continuously update data fetched from a source through a series of operations (e.g., transformation, analysis, etc.). I'm considering two approaches: 1. Using Assets: This involves materializing assets after each operation. 2. Fetching Data, Performing Operations, and Uploading to S3: Where I fetch the data, perform the necessary operations, and upload the results to S3 after each step. I would appreciate your insights on the following: 1. Any recommendations on these approaches based on performance, scalability, or resource usage? 2. Are there better alternatives or methodologies I should consider? 3. Any specific constraints or requirements I should keep in mind for this data processing pipeline (e.g., real-time updates, security concerns, or storage limitations)? Thank you for your help!

Thomas

11/30/2023, 10:50 AM

Hello, I have an external service in REST I want to cosume with dagster. Two use case : • event based : A post request is made with a callback url • Queue based : A post is request. an id for the identificated task. A get request is made on this ID to check if the job is done. How would you handle those kind of tasks ? I don't find any exemple. For queue based, I guess a sensor is OK. but for the 1st one I don't know without creating some middleware

Ben Jeffrey

11/30/2023, 2:22 PM

Hey team, We're currently stringing together multiple jobs but need to track the whole process end-to-end for each materialisation. Currently our jobs are linked by asset sensors and this works great, however we need some sort of ID to be able to query all runs and find the runs responsible for a successful end-to-end materialisation. Put simply say we have: Job A -> (Sensor 1) -> Job B -> (Sensor 2) -> Job C What is the best way for me to link the runs of each job together using a singular ID? So that I can query all runs based on one ID and obtain a set of runs for the job chain? Is it possible to set the run ID of a run in the run_launcher, though I feel like this may not be the best idea? And if that is not advisable, is the next best thing using tags with a specific ID?

Hernan

11/30/2023, 3:26 PM

Hi, I'm trying to submit a databricks job using dagster, I'm getting an unexpected error: module 'databricks.sdk.service.jobs' has no attribute 'JobsHealthRule'

lautaro kogan

11/30/2023, 3:46 PM

Hi, community. I'm trying to force a failure in an asset run whenever a certain criteria is met. The criteria would be something like this

if not os.path.isabs(path)

Jay Spiers

11/30/2023, 6:24 PM

Hello all, I am having trouble when trying to utilize the "Reload definitions" option in the webserver UI. I am using a GRPC server and trying to take advantage of the relatively new command "dagster code-server start" to prevent the need of rebooting the server. When I click Reload it eventually times out after 60 secs despite my configuration setting at 600 secs. Not exactly sure what I'm doing wrong so any help would be much appreciated! Attaching screenshot of config and error message below Thanks Error message:

Copy code

dagster._core.errors.DagsterUserCodeUnreachableError: User code server request timed out due to taking longer than 60 seconds to complete.

  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_core\workspace\context.py", line 608, in _load_location
    origin.reload_location(self.instance) if reload else origin.create_location()
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_core\host_representation\origin.py", line 361, in reload_location
    return GrpcServerCodeLocation(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_core\host_representation\code_location.py", line 629, in __init__
    self._external_repositories_data = sync_get_streaming_external_repositories_data_grpc(
                                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_api\snapshot_repository.py", line 25, in sync_get_streaming_external_repositories_data_grpc
    external_repository_chunks = list(
                                 ^^^^^
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_grpc\client.py", line 346, in streaming_external_repository
    for res in self._streaming_query(
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_grpc\client.py", line 184, in _streaming_query
    self._raise_grpc_exception(
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_grpc\client.py", line 135, in _raise_grpc_exception
    raise DagsterUserCodeUnreachableError(

The above exception was caused by the following exception:
grpc._channel._MultiThreadedRendezvous: <_MultiThreadedRendezvous of RPC that terminated with:
	status = StatusCode.DEADLINE_EXCEEDED
	details = "Deadline Exceeded"
	debug_error_string = "UNKNOWN:Error received from peer ipv4:127.0.0.1:4266 {grpc_message:"Deadline Exceeded", grpc_status:4, created_time:"2023-11-30T18:13:54.0310382+00:00"}"
>

  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_grpc\client.py", line 180, in _streaming_query
    yield from self._get_streaming_response(
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\dagster\_grpc\client.py", line 169, in _get_streaming_response
    yield from getattr(stub, method)(request, metadata=self._metadata, timeout=timeout)
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\grpc\_channel.py", line 541, in __next__
    return self._next()
           ^^^^^^^^^^^^
  File "C:\ProgramData\miniconda3\envs\dagster\Lib\site-packages\grpc\_channel.py", line 967, in _next
    raise self

reza razavipour

11/30/2023, 7:22 PM

Hello all, I have a job scheduled as the following:

from dagster import AssetSelection, define_asset_job

selection = AssetSelection.keys("create_weekly_folder")

# selection = AssetSelection.keys("report_validator").upstream()

create_auction_root_folder = define_asset_job(

name="create_auction_root_folder",

selection=selection,

How do I guard against multiple runs of this job where I don't have access to context: AssetExecutionContext

🤖 1

Jordan Fox

11/30/2023, 8:20 PM

If I want to build a config based approach for loading source assets, ie, have say a postgres database that populates with metadata describing source systems and then have an app where users can config tables they want to load, what's the right dagster approach to then take that table config and load assets. Is it multi-assets?

R Lucas

11/30/2023, 10:05 PM

Hello, I am interested in generating unpartitioned assets based on the last N partitions of a time-partitioned assets (summary from last 90 days of hourly/daily asset as example). I don't see an existing PartitionMapping class for this use-case. One way to tacle this would be to chain a

TimeWindowPartitionMapping

with a

LastPartitionMapping

My concern about this approach is to generate intermediate data only as source for the

LastPartitionMapping

target asset. I would like to know the opinion from Dagster experts which option would be best: • PA A partition mapping class for such use-case is expected soon • Using intermediate asset of these 2 existing PartitionMapping classes would be the best approach • Implementing my own custom PartitionMapping class

Yuan Cheng

11/30/2023, 11:21 PM

Hi guys, we noticed that it takes a long time for asset to get out of "Preparing" status to "Executing", is there a way to optimize it? We are on dagster 1.3.12 . Thanks

dagster bot responded by community 1

🤖 1

Adam

12/01/2023, 3:07 AM

Hi! can anyone help me understand why the lineage chart looks wonky like this on dagster cloud? it looked fine on local

Bolin Zhu

12/01/2023, 6:47 AM

Hi team, I am connecting Dagster web server to two different code locations, say code-location-a, code-location-b.

slack_on_run_failure_sensor

is defined for each of the code locations using

dagster-slack

. I seem to be getting duplicated alerts when a job from code-location-a failed. I received alerts from both slack_on_run_failure sensors (code-location-a and code-location-b). May I ask how can I eradicate this?

dagster bot responded by community 1

Lukas Haberzettl

12/01/2023, 11:00 AM

Hello everyone. I was wondering, is it possible to run one asset that is being materialized in multiple jobs, always with different resources (for each job)? My situation: I have one graph asset consisting of multiple ops. Each of the ops are working with resource called “downloader” which is an interface (abstract configurable resource) for multiple implementations of different downloaders. My idea is to have multiple asset jobs, all of them calling that one asset but always with different downloader class. I know that I could do special ops for each downloader, but since the pipeline template is the same for all of the downloaders, it seems better to have abstraction on the resource level (so in case of a new downloader, the only thing I need to do is to implement the new downloader class and plug it to the new job). I will be glad for any advice, thanks

Casper Weiss Bang

12/01/2023, 12:57 PM

Utilizing the Pythonic ConfigurableRessources and writing tests, i might have hit a wall of sorts, I have a Resource that depends on another:

Copy code

class DatabricksIOManager(ConfigurableIOManager):
    client: UnityCatalogClient

Which i wanted to test by using the

mock_resource

Copy code

manager = DatabricksIOManager(client=ResourceDefinition.mock_resource())

however i get pydantic validation errors

Copy code

Input should be a valid dictionary or instance of UnityCatalogClient [type=model_type, input_value=<dagster._core.definition...bject at 0x7f2b437d3250>, input_type=ResourceDefinition]
    For further information visit <https://errors.pydantic.dev/2.5/v/model_type>

Bolin Zhu

12/01/2023, 2:20 PM

Hi all! Has anyone faced the following issue: • having a dagster webserver connecting to two code locations (one code location with python assets, one code location with dbt assets) • when submitting a job to materialise a list of selected dbt assets, two duplicated jobs are launched

Gary McClure

12/01/2023, 2:22 PM

Hey there, has anyone created an Airbytes connection for mixpanel and then had load_assets_from_airbyte_instance(airbyte_instance) fail because the mixpanel tables have a space in the name?

Jonathan Williams

12/01/2023, 5:52 PM

Hey everyone, I have a cloud Postgresql database resource hosted from GCP. I'm using dagster serverless. I'm trying to figure out how to set up cloning a Postgres Database for branch deployments. Does anybody have any tips regarding that? If I could, I would create ops connected to a dagster_cloud_build_push which would clone and delete the database like the example shows. However the example is in snowflake and I am tied to Postgres which doesnt seem to have easy capability of cloning a database. Any tips or suggestions would be appreciated!

Jordan Fox

12/01/2023, 9:37 PM

I'm using docker with a container for repo and a container for webserver. What's the appropriate setup to mount the code location in the repo and maintain the webserver's access? When I mount the code location, even the repo container can't read the module anymore. ie, ./folder/opt/dagster/apprwx

Amit Taller

12/01/2023, 9:55 PM

Hey, While running a

run_status_sensor

I'm getting an error (not on every run) on the command:

Copy code

materializations_planned = context.instance.get_records_for_run(
    run_id=context.dagster_run.run_id, of_type=DagsterEventType.ASSET_MATERIALIZATION_PLANNED
).records

Any ideas why? Thanks!!

Roy Ammeraal

12/01/2023, 10:37 PM

Does anyone has a concise and clear working example of an ssh ~~task~~ asset in Dagster? It can be as simple as doing a

touch /tmp/helloworld

on a remote host.

Zach

12/01/2023, 11:31 PM

Yet another "can't figure out how to configure log levels" post. I found this discussion and tried it the same thing by adding this config blob to my run config:

Copy code

loggers:
  console:
    config:
      log_level: WARNING

but it didn't produce any noticeable difference in the logs that get emitted to the UI. I've also looked at some of the docs here, but these seem to be focused on configuring logs globally across a deployment. I need to be able to change the log level for a specific job, and leave other jobs with the default. Any suggestions?

Bowen Zhang

12/02/2023, 3:25 PM

Hi there, I have read about Dagster code locations doc, I have the following questions: 1. Does the setup have to be monorepo? Can I set code location to be different Github Repo? 2. If we can set it to be different Github Repo, for each code location: a. how do we configure the K8s worker image in

dagster.yaml

for different code locations (with different docker images) ?