I m trying to split up an asset in a graph into equal parts

Question

I m trying to split up an asset in a graph into equal parts and then run an op on each of those parts However this is currently failing to load due to `InputDefinition` not being iterable What am I doing wrong here Code in thread

Danny Steffy · Answer

``` graph def full scored data set ids to score pd DataFrame trained model xgb XGBClassifier gt pd DataFrame chunk the recruiters to score and score them in batches chunked ids = np array split ids to score 10 fan outs = for chunk in chunked ids chunk df = pd DataFrame from records chunk fan outs append generate score factory ins= ids to score chunk df trained model trained model return append scores fan outs ```

Danny Steffy · Answer

failing on this part `chunked ids = np array split ids to score 10 `

Danny Steffy · Answer

do I need to partition my input asset by an integer somehow

Zach · Answer

you can t really run arbitrary code in a dagster graph definition the code inside a graph definition is compiled into a DAG when Dagster initially loads your code so it doesn t run the way you would expect when you actually run a job to do a fan out you ll want to encapsulate that logic into an <https docs dagster io concepts ops jobs graphs dynamic graphs|op that yields a DynamicOutput> for each chunk you want to analyze in your fan out ops

Danny Steffy · Answer

would changing the graph to a job allow it to compile the way I expect or I would need to create an op factory that would generate a dynamicoutput for each chunk

Zach · Answer

a job is just a configured and executable graph so that won t help you although you ll need to turn it into a job to execute it which you can do simply with <http full scored data set to|full scored data set to> job you shouldn t need an op factory you just need an upstream op that encapsulates all that chunking logic something like ``` op out=DynamicOut def chunking op ids to score trained model chunked ids = np array split ids to score 10 fan outs = for chunk in chunked ids chunk df = pd DataFrame from records chunk yield DynamicOutput ids to score chunk df trained model trained model graph def full scored data set gt pd DataFrame chunk the recruiters to score and score them in batches result = chunking op map generate score factory collect return append scores result ```

Danny Steffy · Answer

ah I see

Danny Steffy · Answer

thank you

Zach · Answer

no problem hope that helps

Danny Steffy · Answer

I m getting this error now `AttributeError OpDefinition object has no attribute map ` Am I missing some sort of import or does map not work for this case

Zach · Answer

how are you executing this code

Danny Steffy · Answer

I m just trying to load it into dagster as a code location

Zach · Answer

ah looks like I missing the to actually call the chunking op ```result = chunking op map generate score factory collect ```

Danny Steffy · Answer

ah I see So then I would need to get rid of my generate score factory which is an op generating factory and instead change that to an op that uses a Dynamic Input

Danny Steffy · Answer

actually it looks like there aren t dynamic inputs I need the scoring op to also take in another input asset besides the chunking op

Danny Steffy · Answer

ah looks like a lambda

Danny Steffy · Answer

I see that in the doc now

Danny Steffy · Answer

oh wait I see the chunking op was already passing that through

Zach · Answer

yeah you shouldn t need to do the op factory thing

Danny Steffy · Answer

how do I access those inputs in my called scoring op though

Danny Steffy · Answer

``` op out=DynamicOut def chunking op recruiter team to score trained model chunked ids = np array split recruiter team to score 10 for chunk in chunked ids chunk df = pd DataFrame from records chunk yield DynamicOutput ids to score chunk df trained model trained model graph def full scored data set gt pd DataFrame chunk the recruiters to score and score them in batches result = chunking op map generate score for ids collect return append scores result op required resource keys= sql 04 resource def generate score for ids context ins Dict str In scoring df = data set to score batch recruiters teams to score=ins ids to score sql conn=context resources sql 04 resource return scored data scoring df ins trained model ```

Zach · Answer

same way you would any input ``` op def scoring op context fanned out input f this is the ids to score fanned out input ids to score f this is the trained model trained model ```

Zach · Answer

yeah that looks right

Danny Steffy · Answer

hm I get an error saying `generate score for ids is not defined`

Zach · Answer

yeah you need to define it above your graph

Zach · Answer

python executes from top down

Zach · Answer

so it s not defined when python reads your graph definition

Danny Steffy · Answer

face palm thanks still pretty new to python

Zach · Answer

no worries I still do that stuff and I ve been coding for like 15 years lol

Danny Steffy · Answer

so at the very end of the file I put in `graph asset = AssetsDefinition from graph full scored data set ` to turn that into an asset right when I try to materialize that I get this error ``` ASSET JOB cannot be executed with the provided config Please fix the following errors Missing required config entry ops at the root Sample config for missing entry ops full scored data set ops chunking op inputs recruiter team to score lt selector gt trained model lt selector gt ``` Do I need to specify those values in the chunking op as specific assets we have defined elsewhere

Zach · Answer

nice so the graph is compiling so I m not sure how well assets and ops interoperate right now so to keep it simple I d start with loading these inputs through an op the job needs direction on where your inputs are stored i e where your dataframe and model live are these stored in a filesystem or on cloud storage somewhere

Zach · Answer

you ll basically want an initial op that can explicitly load those objects from wherever they re stored possibly with the paths configurable via <https docs dagster io concepts configuration config schema run configuration|op config> then pass those objects from that op to your chunking op

Zach · Answer

if those are assets somewhere already defined in Dagster and can be imported into the module maybe this is more where you were headed since you created an asset from the graph then you can make your graph backed asset dependent on them like in <https docs dagster io concepts assets graph backed assets defining basic dependencies for graph backed assets|this example>

Danny Steffy · Answer

ah it looks like adding the keys by input name and passing those through the graph worked and it was able to attempt to run

Danny Steffy · Answer

looks like the DynamicOutput is missing a mapping key though

Zach · Answer

ah yeah good catch

Danny Steffy · Answer

it s running thanks for all your help

Zach · Answer

awesome glad we got it going

Danny Steffy · Answer

so just want to double check with this this will parallelize these and we can have each of the generate score for ids op in their own container in production We re just starting out to use dagster

Zach · Answer

it depends on what executor you re using if you re using the multiprocess executor which is the default it ll launch each one in its own process you can use the <https docs dagster io apidocs libraries dagster docker dagster docker docker executor|docker executor> to launch each step in its own container

Danny Steffy · Answer

excellent thank you

Danny Steffy · Answer

just to follow up on this using the docker executor I have this graph defined as an asset and then have a sensor that uses a `define asset job` to materialize that asset It looks like the entire job is run inside a single docker container How can I specify that each step particularly the fan out should be done in it s own docker container

Danny Steffy · Answer

it looks like `define asset job` takes in a config which I m presuming is where I can specify the executor But I m not sure how to go about doing that

Danny Steffy · Answer

oh wait there s `executor def` facepalm

Danny Steffy · Answer

to come back to this much much later I wrote most of our scoring processing in plain python instead of defining ops for each as I m not sure the best way to write them as ops We now have the keys we want chunked as their own ops so basically batches of 300+ keys in a single op I now want to use those keys to make another database dip get data and run that data through our scoring system and then output that data somewhere My thought is the database dip the scoring and the output could all be their own op with different IO managers and storing the data in memory temporarily until the op is finished But I m not sure how to piece those together in the code I m currently just using a for loop and going over each key in plain python How would it look using ops instead

Danny Steffy · Answer

Would it be as simple as changing the map call here to run multiple ops How could I tell the mapped op that it needs to run the process on each row ``` chunking op recruiter teams to score map lambda batch generate score for ids batch trained model collect ```

Danny Steffy · Answer

I m just not sure how to change this up to be more in line with an op There is where I m doing my for loop for the keys in the batch and reading the data in which we immediately score against a model instead of writing out as an asset or anything ```def score data set batch context recruiters teams to score pd DataFrame sql conn trained model score df = pd DataFrame dirname = os path dirname file filename = os path join dirname sql recruiter user recruitable athlete sql count = 0 tic = dt datetime now logging tic = dt datetime now scoring metadata = for index row in recruiters teams to score iterrows sql = open filename read sql = sql replace RecruiterUserid str row userid sql = sql replace RecruiterTeamid str row teamid start time = time time succeeded = False retry count = 0 scoring batch = None while not succeeded and retry count lt 5 try scoring batch = pd read sql sql sql conn ```

Danny Steffy · Answer

would I maybe need to fan out twice