Hi team! We have a pipeline to grab a lot of paged data, e.g.:

@graph
def get_and_persist_page(page_idx):
  page_data = get_page(page_idx)  # from external API
  persist_original(page_idx, page_data)  # persist to track history 
  return to_dataframe(page_idx, page_data)  # convert to Pandas DataFrame

@graph
def my_job_graph:
  page_indexes = get_page_indexes()  # DynamicOutput, get list of pages from external API
  list_of_dfs = pages.map(get_and_persist_page).collect()
  combined_df = combine(list_of_dfs)
  save_to_warehouse(combined_df)

It works this way. I like that it is possible to configure retry-policy for "get_page" op. But I'm curious if there is another elegant way to do this? What bothers me: • I don't know how to define type for the list of Pandas DataFrames produced by

collect()

• It produces some "noise" in Dagit for my 100 pages. Is there a way to group logs/charts for `DynamicOutput`s somehow?

Mykola, if you're asking about the appropriate type hint for the

combine

function, in my experience

List[pd.DataFrame]

is interpreted correctly and passes dagster's internal type checks. (Not a member of dagster team, just trying to be helpful)

@Dmitry Mogilevsky, thanks, that's probably the best idea in this case! But I was curious about combining Dagster types and List, e.g. type generated by

create_dagster_pandas_dataframe_type()

Ah yes, I haven't gotten around to trying that yet.

hey @Mykola Palamarchuk - you should be able to do something like:

from dagster import op, DagsterType, job, DynamicOut, DynamicOutput, List, In

EvenDagsterType = DagsterType(
    name="EvenDagsterType",
    type_check_fn=lambda _, value: isinstance(value, int) and value % 2 == 0,
)


@op(out=DynamicOut())
def start():
    yield DynamicOutput(2, mapping_key="a")
    yield DynamicOutput(4, mapping_key="b")


@op
def middle(x):
    return x


@op(ins={"x": In(dagster_type=List[EvenDagsterType])})
def combine(x):
    ...


@job
def my_job():
    combine(start().map(middle).collect())

@sandy, thanks! 1. I've just checked

In(dagster_type=List[SomeDagsterType])

. I'm having this error:

UserWarning: Error loading repository location repo.py:TypeError: Parameters to generic types must be types. Got <dagster.core.types.dagster_type.DagsterType object at 0x7f4994fff490>.

I'm using Dagster type generated by

create_dagster_pandas_dataframe_type()

2.

Do you have thoughts on what this would ideally look like?

I'd like there was a possibility to fold/unfold timeline view for dynamic graphs somehow. It does not make a lot of sense to show them all when running on a large sets of chunks.

is the

List

in your example imported from the

typing

module? the code I pasted will only work if you're using

List

that's imported from Dagster. also, what version of Python are you doing?

Oh, that works with

List

from Dagster module! Thanks!

thanks for the suggestions @Mykola Palamarchuk. I filed a couple of issues to track them: • https://github.com/dagster-io/dagster/issues/8491 • https://github.com/dagster-io/dagster/issues/8492