makes sense

One of our focuses this quarter is to really nail this devex where people essentially want to a pipeline to a certain state and then iterate on one task

So we are cooking up the notion of grouping of runs/build/session where all runs within it would share intermediates

(with user control over invalidation)

perfect -- that's exactly along the lines of what I was thinking.

Although the error explanation is quite specific, I can’t make it to work. How can I get around:

dagster.core.errors.DagsterInvalidDefinitionError: In @pipeline networkscore_incremental_pipeline received invalid type <class 'dagster.core.definitions.composition.CallableSolidNode'> for input "tag_session" (at position 0) in solid invocation "update_tags". Must pass the output from previous solid invocations or inputs to the composition function as inputs when invoking solids during composition.

My pipeline:

@pipeline(...)
def networkscore_incremental_pipeline():
    df_tag_session = load_delta_table_to_df.alias("tag_session_load_delta_table_to_df")
    df = update_tags(df_tag_session, sa_simulation_of_input_nps_categories(),)

I found that the error is in

load_delta_table_to_df

, if I use something different it works. I figured maybe the return of the solid is not typed as

DataFrame

. Tried _`output_defs`_ and other things. What I’m missing? my very easy update_tags-solid:

@solid(
    description='''Update Tag to Session mapping table with new inputs from external applications (e.g. SA)''',
)
def update_tags(context, tag_session: DataFrame, new_tags_sessions: DataFrame) -> DataFrame:
    return tag_session

and my error source in

load_delta_table_to_df

, but why?

@solid(
    required_resource_keys={'spark', 's3'},
    description='''Loads given delta coordinates into a spark data frame''',
    # output_defs=[OutputDefinition(name='df', dagster_type=DataFrame, is_optional=False),],
)
def load_delta_table_to_df(
    context, delta_coordinate: DeltaCoordinate, where_conditions: String,
) -> DataFrame:

    delta_path = get_s3a_path(
        delta_coordinate['s3_coordinate_bucket'], delta_coordinate['s3_coordinate_key']
    )
    df = (
        context.resources.spark.spark_session.read.format("delta")
        .load(delta_path)
        .where(where_conditions)
    )
    return df

Thanks a lot for any help, hint or others.

Instead of

df_tag_session = load_delta_table_to_df.alias("tag_session_load_delta_table_to_df")

you need

df_tag_session = load_delta_table_to_df.alias("tag_session_load_delta_table_to_df")()

the alias acts just like the original solid, so one has to invoke it

The error message is quite the mouthful for sure

Tracking issue for better message here: https://github.com/dagster-io/dagster/issues/2308

Hello, when i have solids creating database objects (tables, views) and other solids depending on those objects, how do i model the dependencies ? should i use outputs and inputs ? or materialisations ? (but one can't depend on materialisations ? right ?)

actually i don't really depend on the table but on the data in it, but the data is too big to be passed as a pandas dataframe.

hi, i'm getting the error:

pipeline_1  | Error: A scheduler must be configured to run schedule commands.
pipeline_1  | You can configure a scheduler on your instance using dagster.yaml.
pipeline_1  | For more information, see:

even thought i have a scheduler defined in dagster.yaml:

scheduler:
  module: dagster_cron.cron_scheduler
  class: SystemCronScheduler

Resolved the above, now getting:

dagster.check.CheckError: (<class 'dagster.core.launcher.RunLauncher'>, 'class RunLauncher in module dagster.core.launcher', <class 'dagster.core.serdes.ConfigurableClass'>)

my

dagster.yaml

is:

run_launcher:
  module: dagster.core.launcher
  class: RunLauncher

it'd be nice to have some opencensus integration in dagster -- both having dagster emit traces+metrics but also perhaps a helper to thread trace context via the dagster context

I found some old conversation somewhat related, but couldn't find anything conclusive. Even if a solid inputs is marked optional my execution fails unless I at least configure an empty input with that name. Is this expected behavior?

is it possible to use ipdb to debug solids?

have not had any luck getting it to work

has there been any discussion regarding some sort of hashing of the code in a solid -- it'd be nice for a subsequent run to (optionally?) invalidate the intermediates of solids that have had code changes

I've run in to some trouble using the dagster-pandas DataFrame validation. The builtin ColumnTypeConstraints are limited in the dtypes they recognize. For example the string_column only checks for 'object' and not the Pandas 'string' dtype, boolean_cooumn only checks 'bool' and not the Pandas 'boolean' dtype, etc. Would there be any interest in reworking the ColumnTypeConstraints to use the built in Pandas dtype checking API (i.e. is_string_dtype)?

Issue for tracking here: https://github.com/dagster-io/dagster/issues/2317

is it possible for a downstream solid to run even if one of the upstream solids is conditional doesn't run?

eg, have two solids feed into one solid, only one of the two runs, downstream solid conditionally receives input from one of them and runs

is it possible to get instance object from solid?

oh ok i got the currence instance object by

instance = DagsterInstance.get()

Hi again guys, I've got a small problem with the timestamps on dagit runs/logs

Is it possible for resources to reference/depend on other resources?

Hey quick question—what is the best practice for creating a dependency structure/connecting solids together within a pipeline when the solids don't actually have values being passed between them--for example, I have a "load" solid that loads my data into a DB table, and then a "transform" solid to transform the data in that table once the load solid is finished. What is the best practice for creating this dependency if no data is actually being passed directly from the "load" solid to the "transform" solid? Thanks!

we have a "Nothing" type