onTrigger is executed and passed processor context and session
(context, session)
| 62 | return len(self.content) |
| 63 | |
| 64 | def onTrigger(context, session): |
| 65 | """ onTrigger is executed and passed processor context and session |
| 66 | """ |
| 67 | flow_file = session.get() |
| 68 | if flow_file is not None: |
| 69 | read_cb = ContentExtract() |
| 70 | # read flow file content into read_cb.content data member |
| 71 | session.read(flow_file, read_cb) |
| 72 | # create empty in-memory text streams csv_data buffer |
| 73 | csv_data = StringIO() |
| 74 | # load str data into datatable, then convert to pandas df |
| 75 | dt_frame = dt.Frame(read_cb.content) |
| 76 | pd_dframe = dt_frame.to_pandas() |
| 77 | # convert df to csv file like object without df index |
| 78 | pd_dframe.to_csv(csv_data, index=False) |
| 79 | # set the csv to the start of text stream |
| 80 | csv_data.seek(0) |
| 81 | # get csv str data out of StringIO text stream, then store str |
| 82 | csv_data = csv_data.read() |
| 83 | # write csv str to flow file |
| 84 | write_cb = ContentWrite(csv_data) |
| 85 | session.write(flow_file, write_cb) |
| 86 | session.transfer(flow_file, REL_SUCCESS) |
nothing calls this directly
no test coverage detected