Hello, how can I use Pyspark alongside dagster? Is there any practice like deploying Pyspark code as API and executing it from dagster?

Hello 👋 I want to use Dagster Cloud Serverless, but I don't know how to approach the following 2 use cases 1. For the first use case, we need to run (and interact with) a dockerized MSSQL-server instance. a. Is it possible, from a job in a Dagster Serverless environment, to run Docker or Docker Compose? 2. For the second use case, we need to access an API on a private network, which is only accessible if we can provide IP addresses to be whitelisted. a. Would we be able to get (a range of) IP addresses, from which our jobs will be executed? b. Alternatively, would we be able to use a VPN (e.g. OpenVPN), within the Dagster Serverless environment?

Hi! I would like to get the list of the keys of the materialized and non materialized partitions from a notebook, is there any possibility to do that?

here is my current config, for easy copy/pasting and also searchability:

ops:
  execute_mi_sales_zip_30_lag_est_snowpark:
    config: {'inputs': {'models_ran': True}}
  run_metrics_models:
    config: {'inputs': {'lag_365_est': True, 'lag_90_est': True}}

What parameters do I need to set for Github?

Hi CJ - if you set up the Github Action to enable CI/CD for Dagster Cloud, this will be managed automatically on your behalf. You can check out the github action and the quick start repositories here: https://github.com/dagster-io/dagster-cloud-action/

Is it possible for an op to introspect on the

Executor

being used by its job? Context: I am trying to construct an op with the following properties: 1. Uses a

DynamicOut

2. The

DynamicOut

produces exactly one output if running in a job whose

Executor

is an

InProcessExecutor

3. If the job’s

Executor

is a

MultiprocessExecutor

, the `DynamicOut`’s branching factor matches the

max_concurrent

of the executor The overall goal is to perform batch processing split amongst a number of processes that matches the MP factor of the job. We haven’t had success with allowing Dagster to control this itself using a

DynamicOut

with arbitrary fanout, because the time needed to spawn one new process per item is unacceptably high even when we use `forkserver`.

Hi there, I'm wondering about a "best practice" scenario using sensors. My situation... • I plan on using nats.io + jetstream to allow for durable subscriptions. • I'd like each repository / sensor to be able to self manage their subscriptions (subscription id, topic patterns, etc...) • Then in my sensors I want to pull messages for those subscriptions. The thing I'd like to optimize is that call to create or update the subscription. What's the best way to implement that to reduce how frequently it is called? Ideally once per user deployment of the repo IMO. Thanks!

Hi Team — Here’s my situation: •

Pipeline Job 1

is scheduled to run @ 4am Daily (using schedule) •

Pipeline Job 1

materializes an asset (using AssetMaterialization on the last Op in the job) ◦ Asset only needs to be materialized on the first of the month, but I’d still like

Pipeline Job 1

to run daily •

Pipeline Job 2

sensing when to run AFTER

Pipeline Job 1

materializes an asset on the first of the month (using asset sensor) How do I use partitions to only materializes

Pipeline Job 1

’s asset on the first of the month? Can you share pseudo-code on how to achieve this?

Hi dagster people, I have made a resource that contains a class which just initiate an object that can post an api request to a Hightouch endpoint. I have a op that reference the resource with

required_resource_keys={"hightouch"}

and a job that call the op with

_resource_defs_={_"resources"_:{_"hightouch"_: hightouch_resource}}

But I got an error like this , what might be the issue, is it syntax issue?

dagster._check.CheckError: Value in Mapping mismatches expected type for key resources. Expected value of type <class 'dagster._core.definitions.resource_definition.ResourceDefinition'>. Got value {'hightouch': <dagster._core.definitions.resource_definition.ResourceDefinition object at 0x146eb8c40>} of type <class 'dict'>.

# My resources
class Hightouch:
    _endpoint = "some endpoint"
    def __init__(self, api_key: str):
        self,
        self.api_key = api_key
   
    @property
    def headers(self):
        return {"Authorization": f"Bearer {self.api_key}"}

    def start_sync(self, sync_id: int) -> int:

        response = <http://requests.post|requests.post>(
            self._endpoint + f"{sync_id}/trigger", headers=self.headers
        )
        return response

@resource(
    config_schema={
        "HIGHTOUCH_API_KEY": Field(StringSource, is_required=True),
    }
)
def hightouch_resource(context) -> Hightouch:
    return Hightouch(
        api_key=context.resource_config["HIGHTOUCH_API_KEY"],
    )
hightouch_configured = hightouch_resource.configured(
    {
        "HIGHTOUCH_API_KEY": os.getenv("HIGHTOUCH_API_KEY"),
    }
)

# My ops
from resources.hightouch import *

@op(required_resource_keys={"hightouch"})
def sync_user(context):
    """
    This sync function only syncs user on Hightouch
    """
    syncs = context.resources.hightouch
    syncs.start_sync()

#My jobs
@job(
    description="""
        This job runs the op
    """,
    tags={"ecs/cpu": "2048", "ecs/memory": "8192"},
    resource_defs={"resources":{"hightouch": hightouch_resource}},
    config=some_config,
    
)
def my_job():
    sync_user()

Hi dagster folks. I’m running into an issue where I’m able to run a graph locally, but when I push to production and run it in Dagster Cloud I get Multiprocess executor error.

Multiprocess executor: child process for step get_community_updates unexpectedly exited with code -11
dagster._core.executor.child_process_executor.ChildProcessCrashException

Stack Trace:
  File "/usr/local/lib/python3.8/site-packages/dagster/_core/executor/multiprocess.py", line 214, in execute
    event_or_none = next(step_iter)
,  File "/usr/local/lib/python3.8/site-packages/dagster/_core/executor/multiprocess.py", line 330, in execute_step_out_of_process
    for ret in execute_child_process_command(multiproc_ctx, command):
,  File "/usr/local/lib/python3.8/site-packages/dagster/_core/executor/child_process_executor.py", line 163, in execute_child_process_command
    raise ChildProcessCrashException(exit_code=process.exitcode)

the

get_community_updates

is pulling down a few <100 rows from snowflake and trying to return a pandas dataframe with snowflakes.connector and the fetch_pandas_all() function. I’ve tried increasing the ecs/cpu tags but no luck there. Any tips/pointers would be appreciated

Using

op_tags

on

@asset

to configure the k8s job is not working for me, but from looking at the code/docs, it seems like it should be combining anything in the

pod_spec_config

key in

"dagster-k8s/config"

w/ the dagster-generated job/pod config. I'm using the helm chart to install dagster (w/ run launcher

K8sRunLauncher

) + the user-deployment helm chart to keep user code in a separate repo/namespace.

@asset(
	op_tags={
		"dagster-k8s/config": {
			"pod_spec_config": {
                "tolerations": [{
				    "key": "my-key",
				    "operator": "Exists",
				    "effect": "NoSchedule",
			    }],
            },
		},
	},
)
def my_asset(_):
	pass

Is there a way to rename a given required resource required by some io_manager / op? For example, I have a few ops and io_managers that require a certain resource and use the

required_resource_keys={…}

syntax, here I’ve used generic names. I’d like to be able to override these in certain cases: eg if I have a single job that may need multiple of that type. Is there some way to provide a mapping in these cases? EG:

@io_manager(required_resource={"db_creds"})
def my_io_manager(context): ...

resource_defs_for_job = {"my_first_db_creds":db_creds(...), "my_2nd_db_creds":db_creds(...), "my_first_io_manager":my_io_manager, "my_2nd_io_manager":my_io_manager}

How can I force

my_2nd_io_manager

to use the required resource key

my_2nd_db_creds

instead of

db_creds

?

Should

multi_asset_sensor

work with assets across multiple repositories? The docs for regular asset sensors suggest that is allowed, so I was hoping it'd work with multi_asset_sensors as well. However, I'm getting a grpc error from dagster-daemon:

details = "Exception iterating responses: No asset with AssetKey(...) found in repository"

. The asset key in question is defined by another repository.

Does the reload repository on the bottom left corner of dagit UI work if running dagster through docker compose (like this example)? I tried to volume mount my code instead but reload repository doesn’t work here as well. Further, in the latter case, since the user code had been volume mounted in the docker user code container, it couldn’t run each step in a new container because the new container didn’t have the volume mounted code. Any solutions? Thanks!

Hello, everyone. Does anyone experience in attaching remote debugger to dagit? What I want is to setup debugger the way, I can execute my job from dagit/dagster cli and setup breakpoints in PyCharm. So, when I start the job in dagit, I can debug the code in Pycharm(go through execution step by step). I know I could do similar things by debugging unit-tests, as dagster jobs and assets, are just simple functions. But I would like to debug the complete pipeline.

Can we hide some buttons in the Dagster log page is it possible?

Is it possible to attach dynamic metadata to SDAs? I have an asset:

@asset(config_schema={"thing" = Field(str)})
def my_asset(context):
  thing = context.op_config["thing"]
  ...

How can I attach the contents of "thing" to the asset as metadata?

Hey all - is there an out of the box way to add tags defined in DBT to their representative assets in Dagster? Or I could be barking up the wrong tree here. What I'm trying to test is if I can limit the concurrency of certain DBT models based on a defined tag they might have

Hello there 🙂 I’m working for Weights &amp; Biases. We have built a Dagster integration to help the users we have in common. We have been using it internally and would love to open source it. Would it be possible to discuss a pull request on your repo and work with your team to promote this integration? The integrations helps to: • use and create W&B Artifacts • run training jobs on dedicated compute using W&B Launch • use and create Registered Model in W&B Model Registry • use the wandb client in ops and assets

Hello! I'm running a job a fairly larger scale than before with some ops taking >1 hr to run, which is fine. However for some other faster

op

in the same

graph

, I see them "restarting" by themselves (see logs) about 4 hours after. I also see them duplicated on the graph view, which was not the case before and is, in no way, part of the graph logic. Is there some sort of hidden retry logic that may trigger this? This is a problem because the logic in the

ops

is not meant to be run twice

Hi team! I am running a Dagster process in Dagster Cloud that streams and processes 3 large files from S3 (processing it line by line). The processing is simple, but the files are 1-2 GB in size. When I run this Dagster process in AWS I don't run into any issues. When running in Dagster Cloud I am getting this error: The above exception was caused by the following exception: botocore.exceptions.ResponseStreamingError: An error occurred while reading from response stream: ("Connection broken: ConnectionResetError(104, 'Connection reset by peer')", ConnectionResetError(104, 'Connection reset by peer'))

I know about the limitations to 100 GB a day, but I am very far from that limit today

Have you seen this issue before?

This message was deleted.

Hi team what is the correct way to pass upstream asset key to downstream asset, given the

parents_asset

is listed with

asset_key

array. I want to pass the key to downstream’s test() function. but

parents_asset.asset_key[-1]

does not seem like the right syntax

@asset(

    required_resource_keys={"some_check"},
    key_prefix=DEFAULT_KEY_PREFIX,
    ins={
        "parents_asset": AssetIn(
            asset_key=[
                "snowflake",
                DBT_DATABASE.lower(),
                "my_schema_name",
                "my_table_name",
            ],
            input_manager_key='snowpark_dataframe'
        )
    },

)
def my_test(context, parents_asset):
    context.resources.some_check.test(parents_asset.asset_key[-1])

Hi 👋 I have a partitioned job with two assets. The first asset retrieves a file to eventually split it into several if it is too big. The second asset performs a series of processes, however I would like each sub-file to be processed separately (for memory issues) so in separate runs. I think the most obvious solution would be to split this job into two jobs and add a sensor to detect if there is one or more sub-files per partition to know the number of RunRequest to trigger. However, in my use it is rare that there are several files so I think it's a shame to have to separate the flow of these two assets. I wonder if it is not possible in the io_manager between the two assets, to take the first sub-file for processing (most of the time the only file) and for all other files to trigger a RunRequest on the second asset with the path of this sub-file as configuration. But I don't have the impression that this is feasible. Do you have any suggestions/solutions to achieve this? Thanks in advance

With the databricks step launcher I think I noticed that each asset would create a new job cluster. What would be the best to bundle up all my assets to execute in one job cluster?