streamcorpus_pipeline/_pipeline.py
def _process_output_chunk(self, start_count, next_idx, sources, i_str,
                              t_path):
        '''
        for the current output chunk (which should be open):
          1. run batch transforms
          2. run post-batch incremental transforms
          3. run 'writers' to load-out the data to files or other storage
        return list of paths that writers wrote to
        '''
        if not self.t_chunk:
            # nothing to do
            return []
        self.t_chunk.close()

        # gather the paths as the writers run
        o_paths = None
        if len(self.t_chunk) > 0:
            # only batch transform and load if the chunk
            # isn't empty, which can happen when filtering
            # with stages like "find"

            # batch transforms act on the whole chunk in-place
            logger.info('running batch transforms on %d StreamItems',
                        len(self.t_chunk))
            self._run_batch_transforms(t_path)

            self._maybe_run_post_batch_incremental_transforms(t_path)

            # only proceed if above transforms left us with something
            if (self.t_chunk) and (len(self.t_chunk) >= 0):
                o_paths = self._run_writers(start_count, next_idx, sources,
                                            i_str, t_path)

        # we're now officially done with the chunk
        self.t_chunk = None

        # If we wrote some paths, update the data dictionary of outputs
        if self.work_unit and o_paths:
            old_o_paths = self.work_unit.data.get('output', [])
            o_paths = old_o_paths + o_paths
            self.work_unit.data['start_count'] = next_idx
            self.work_unit.data['output'] = o_paths
            self.work_unit.update()
def _process_output_chunk(self, start_count, next_idx, sources, i_str,
                              t_path):
        '''
        for the current output chunk (which should be open):
          1. run batch transforms
          2. run post-batch incremental transforms
          3. run 'writers' to load-out the data to files or other storage
        return list of paths that writers wrote to
        '''
        if not self.t_chunk:
            # nothing to do
            return []
        self.t_chunk.close()

        # gather the paths as the writers run
        o_paths = None
        if len(self.t_chunk) > 0:
            # only batch transform and load if the chunk
            # isn't empty, which can happen when filtering
            # with stages like "find"

            # batch transforms act on the whole chunk in-place
            logger.info('running batch transforms on %d StreamItems',
                        len(self.t_chunk))
            self._run_batch_transforms(t_path)

            self._maybe_run_post_batch_incremental_transforms(t_path)

            # only proceed if above transforms left us with something
            if (self.t_chunk) and (len(self.t_chunk) >= 0):
                o_paths = self._run_writers(start_count, next_idx, sources,
                                            i_str, t_path)

        # we're now officially done with the chunk
        self.t_chunk = None

        # If we wrote some paths, update the data dictionary of outputs
        if self.work_unit and o_paths:
            old_o_paths = self.work_unit.data.get('output', [])
            o_paths = old_o_paths + o_paths
            self.work_unit.data['start_count'] = next_idx
            self.work_unit.data['output'] = o_paths
            self.work_unit.update()
def _process_output_chunk ( self , start_count , next_idx , sources , i_str , t_path ) : if not self . t_chunk : # nothing to do return [ ] self . t_chunk . close ( ) # gather the paths as the writers run o_paths = None if len ( self . t_chunk ) > 0 : # only batch transform and load if the chunk # isn't empty, which can happen when filtering # with stages like "find" # batch transforms act on the whole chunk in-place logger . info ( 'running batch transforms on %d StreamItems' , len ( self . t_chunk ) ) self . _run_batch_transforms ( t_path ) self . _maybe_run_post_batch_incremental_transforms ( t_path ) # only proceed if above transforms left us with something if ( self . t_chunk ) and ( len ( self . t_chunk ) >= 0 ) : o_paths = self . _run_writers ( start_count , next_idx , sources , i_str , t_path ) # we're now officially done with the chunk self . t_chunk = None # If we wrote some paths, update the data dictionary of outputs if self . work_unit and o_paths : old_o_paths = self . work_unit . data . get ( 'output' , [ ] ) o_paths = old_o_paths + o_paths self . work_unit . data [ 'start_count' ] = next_idx self . work_unit . data [ 'output' ] = o_paths self . work_unit . update ( )
for the current output chunk (which should be open):
          1. run batch transforms
          2. run post-batch incremental transforms
          3. run 'writers' to load-out the data to files or other storage
        return list of paths that writers wrote to
for the current output chunk ( which should be open ) : 1 . run batch transforms 2 . run post - batch incremental transforms 3 . run writers to load - out the data to files or other storage return list of paths that writers wrote to