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()
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