it is possible to create an AssetMaterialization in a create_databricks_job_solid that represent a table processed in a databricks spark_jar_task ?

question about using env vars in the resource config in the run config when using

execute_solid

- I tried to pass in a database password like

'password': {'env': 'DB_PASSWORD'}

but it seems like that isn’t parsed properly - when I inspect

context.resource_config

within the resource it just returns the literal dictionary. Is this expected behavior? Should I just use

os.environ

? it’s confusing since it works for solid configs

get_s3_keys

api has only implementation for

last_run_key

not for

last_completion_time

, Am I right?

is it possible for Asset Sensors to match on wildcard assetkeys? e.g.

@asset_sensor(asset_key=AssetKey(["sftp", "*"]), pipeline_name="process_file_pipeline")

Quick question to `execute_solid()`: Shouldn't that take

input_values

or

run_config

parameter for my solid inputs? I wasn't able to make it work. What finally worked if I directly submitted it to the solid itself (see example in thread).

I've a pipleline with 3 solids which is very basic and I'm triggering it using python api by calling the execute_pipeline in same py file. Now the question is do I really need dagit or k8s operator/daemon to deploy it or I could create a simple python container which executes the py file with dagster dependency. Note: I'm really new and desperate to use dagster and didn't understand how it's being used in production.

How “experimental” are the new core APIs (graphs/jobs/ops)? We’re just integrating Dagster into our infrastructure and are wondering if we should start to use the 1.0 abstractions right off the bat. 🤔

Can I “refresh” an asset so that an

@asset_sensor

sensor triggers to run again? I don’t want to have to delete/wipe the asset in dagit and then have to run a prior workflow to regenerate the asset. I want to kick off the sensor rather than the graph/pipeline that it yields a

RunRequest

for because the sensor uses asset_event metadata to set the run_config for the RunRequest.

can I query for asset materializations by asset key from a sensor evaluation context? I want to do a “negative” asset sensor, i.e. I want to sense when an asset doesn’t exist yet, and if so, yield a run request.

Hello. Im having problems to deploy dagster on an EC2 server with ubuntu. I'm using kubectl, minikube, helm and when I check the pods. I have the next result.

NAME                               READY  STATUS       RESTARTS  AGE
dagster-daemon-f8656bc5c-4sflz                  0/1   CrashLoopBackOff  1     9s
dagster-dagit-6b85ddd774-qbxc8                  0/1   CrashLoopBackOff  1     9s
dagster-dagster-user-deployments-k8s-example-user-code-1-65wtgz  0/1   CrashLoopBackOff  1     9s
dagster-postgresql-0                       1/1   Running      0     84s

And cannot access to the dagster environment. When I try to deploy without kubernets I can expose it into the port :3000 I'm not sure what is the problem and I can't found something on the web to solve it...

Hi All , anyone knows when the Dagster Cloud is coming out

Hi, I am pretty new to dagster but I am not sure what is going on with this error... Does anyone have insight?

TypeError: unsupported operand type(s) for -: 'tuple' and 'int'
...
 File "C:\Users\...\python\drc-fire\fire_module.py", line 53, in prepare_script1
    start = self.analysisYear - self.baselineLength

analysisyear is defined as an int in config and is an int, and baselineLength is an int. The script/module runs on it own without error and nothing is in a tuple

params class

self.baselineLength =  3

module

start = self.analysisYear - self.baselineLength

dagster

@solid(config_schema={"year":int,
    "region":str,
    "cover":str,
    "covername":str,
    "test":bool})
def run_prep_step_1(context):
    year=context.solid_config['year'],
    region=ee.FeatureCollection(context.solid_config['region']),
    cover=ee.Image(context.solid_config['cover']),
    covername=context.solid_config['covername']
    a = step1(year,region,cover,covername)
    prep_lc = a.prepare_script1()

Ah that's dumb, it was because I am neglectful and left all those commas in my function. Thanks all 😅

Hi all dagster mask I am currently facing a blocker in Dagster with Spark I am not sure what might be the problem since I am pretty new to spark and related environment and but I suspect that IO Management might be it (the data type of output which one solid passes to the other which is a Dataframe in my case). The program works fine when I use an in-process executor, but the same program faces issues once I move to a multi-process executor and I want to have concurrency in my program. So, that is also a confusion since this doesn’t fit with the IO manager story. I am using a docker container for the spark master/worker nodes. 1. What possibly I should do to solve this? (I have attached the code in this thread) 2. While passing-on Dataframe from one solid to another, it is possible to pass it in a streaming/micro-batch fashion and not in a batch system so the next solid doesn’t has to wait for it’s parent solid’s execution to complete? ---------- I read this message by @sandy: “If you’re using a multiprocess executor, you are essentially stuck with saving all the intermediate results on S3 (or other persistent storage), because Spark provides no way to serialize a DataFrame object without serializing out its full contents.” But, I and my team intend to build a platform which is flexible with the customer’s cloud storage availability so sticking to just S3 is not feasible for us (since for example, the user might be having google cloud storage and not S3) I know this is a lot at once but even partial help on any of these questions would be greatly appreciated as the entire framework is starting to get a bit overwhelming, based on the more and more complex infrastructure requirements from our core developer and DevOps team.

Hi all Very much a noob problem but I have deployed successfully dagster to k8s with helm and now I am a bit stuck. I have run the tutorial _hello_cereal_ on my local machine. But now I would like to do the same but on dagster deployed on k8s. Did I miss a k8s _hello_cereal_ somewhere in the doc for that purpose?

Hello, I am having an issue where a Dagster pipeline is failing with the error "dagster.core.errors.DagsterInvariantViolationError: No pipelines or repositories found in "ftp_dag_repo"." This pipeline runs a few hundred times per day, and usually about once or twice per day the pipeline fails with this error. What is causing this error to happen on 1% of the runs while being fine for all the others, and is there a fix for this?

Hi team, I ran into this issue with Dagster when I try to create a directory using os.mkdir and write files into it in a solid. The error says “OSError: Read-only file system” . What I am trying to do is to write some files into one directory, upload that directory to my storage, and then delete the local directory. What should be the best way to do it? Thanks!

Hi team. I was looking in the documentation if there is one option to allow the parameters in config_schema be different type like int or float. I try using Union but I got this error: Thank you in advance

Hi Team, I was trying out the schedules and made an hourly_schedule just to automate a cron task and this poped up. Though the docs are great, I can't find a clear insight on this issue. Any kind of help will do. Thanks in advance.

Error1: dagster.core.errors.ScheduleExecutionError: Error occurred during the execution of run_config_fn for schedule data_fetch_schedule

The above exception was caused by the following exception:

Error2: TypeError: data_fetch_schedule() takes 0 positional arguments but 1 was given

edit: typo

Hi there! Has anyone tried setting up Dagster with CockroachDB for storage?

Hi, hope you are doing great! I'm facing some problems deploying an already made pipeline, when running in dagit. Where do I define the preset? where can I find some tutorials that could guide me from beginner to advance in this topics?

Hi all, I am wondering if there is an API for loading Python modules as solids. I would like to write one solid-per-file and I would like to do it without needing to define a top-level compute function in the file. I know how to achieve what I want using some custom code that builds the solid from module contents using

SolidDefinition

, but is there any kind of official API (or one planned?). Thanks.

Guys, I'm ready to convert solids and pipelines to ops and graphs. The question is, when will the UI honor those new terms? (Using op/graph instead of solid/pipeline)

Heyy! I recently got a new laptop so I installed dagster again on a virtual environment, yet I’m suddenly getting an error. I tested the quickstart code and it still shows

ModuleNotFoundError: No module named 'dagster_cron'

dagster.check.CheckError: Failure condition: Couldn't import module dagster_cron.cron_scheduler when attempting to load the configurable class dagster_cron.cron_scheduler.SystemCronScheduler

I tried running it through dagit and dagster pipeline and creating a new virtual environment. I use python 3.7.7. Would love to have any help Thanks!

Hi, I am trying to run one code sample contains DynamicOutput usage. In the first sample, number_generator yield one number, and the code works properly. However, the second code sample gives an error. The only difference is number_generator yields two numbers (I found that usage in slack messages works properly in other samples.) Given error message is shown below. "dagster.check.CheckError: Invariant failed. Description: unexpected output name in dynamic mapped step resolution" Is there a point I missed during return multiple DynamicOutput? Sample-1

@solid(
 output_defs=[DynamicOutputDefinition(name="num1", dagster_type=int)]
)
def number_generator(_):
 for i in range(10):
   yield DynamicOutput(output_name="num1", value=i, mapping_key=f"deneme_{i}_1")


@solid(
 output_defs=[OutputDefinition(name="power", dagster_type=int)]
)
def return_power(_):
 return 2


@solid(
 input_defs=[InputDefinition(name="num", dagster_type=int),
       InputDefinition(name="power", dagster_type=int)],
 output_defs=[OutputDefinition(name="num_power", dagster_type=int)]
)
def num_powers(_, num, power):
 return num ** power


@solid()
def print_num(_, num):
 print(num)


@pipeline()
def dynamic_pipeline():
 numbers = number_generator()
 power = return_power()
 num_power = numbers.map(lambda num: num_powers(num, power)).collect()
 print_num(num_power)

Sample-2

@solid(
 output_defs=[DynamicOutputDefinition(name="num1", dagster_type=int),
        DynamicOutputDefinition(name="num2", dagster_type=int)]
)
def number_generator(_):
 for i in range(10):
   yield DynamicOutput(output_name="num1", value=i, mapping_key=f"deneme_{i}_1")
   yield DynamicOutput(output_name="num2", value=i, mapping_key=f"deneme_{i}_2")


@solid(
 output_defs=[OutputDefinition(name="power", dagster_type=int)]
)
def return_power(_):
 return 2


@solid(
 input_defs=[InputDefinition(name="num", dagster_type=int),
       InputDefinition(name="power", dagster_type=int)],
 output_defs=[OutputDefinition(name="num_power", dagster_type=int)]
)
def num_powers(_, num, power):
 return num ** power


@solid()
def print_num(_, num):
 print(num)


@pipeline()
def dynamic_pipeline():
 numbers, _ = number_generator()
 power = return_power()
 num_power = numbers.map(lambda num: num_powers(num, power)).collect()
 print_num(num_power)

Hi All, Right now I have two solids to hit a URL and return the response, It's actually same code blocks with different naming since I'm using execute_pipeline to run the Dagster from python, I'm passing two different URL's from the run_config. So the question is, Is there a way to avoid these code duplications to perform same tasks(solids) with different configs in a pipeline? I might reuse the same like 5-10 times in a complex scenario. TIA.

Hi, Just curious, is there any way using which once can easily create repository dynamically in Dagster?

Quick question, and I anticipate a simple answer! How can I define in the

config_schema

a list of one or more Shapes? I have Dagster integrated with a Django application (although I'm really only using the Django ORM). I need to perform a simple Django ORM query using one or more Django ORM filters. I'm thinking:

Shape(
    fields={
        'operator': Selector( ... ), # and vs. or
        'tags': Field(config=list, ...),
        'slice_begin': Field(config=int, ...),
        'slice_end': Field(config=int, ...),
    },
)

But then... can I define a list of Shapes?

Morning guys, Been fun working with Dagster so far. Appreciate all the works y’all have put into it. Just had a quick question regarding pipeline execution timestamps. In airflow, one could use the templating feature like so

"{{ ds }}"

to get the execution timestamp of a DAG. The only way I could come up w/ to something similar was with the below code which requires run storage. I’m curious if there is a simpler way to obtain this information?

from dagster import pipeline, repository, schedule, solid
from dagster.core.storage.pipeline_run import PipelineRunsFilter


@solid
def hello(context):
    filter = PipelineRunsFilter(run_ids=[context.run_id])
    run_record = context.get_step_execution_context().instance.run_storage.get_run_records(filters=filter)