streamcorpus_pipeline/_pipeline.py
def _init_all_stages(self, config):
        '''Create stages that are used for the pipeline.

        :param dict config: `streamcorpus_pipeline` configuration
        :return: tuple of (reader, incremental transforms, batch
          transforms, post-batch incremental transforms, writers,
          temporary directory)

        '''
        reader = self._init_stage(config, 'reader')
        incremental_transforms = self._init_stages(
            config, 'incremental_transforms')
        batch_transforms = self._init_stages(config, 'batch_transforms')
        post_batch_incremental_transforms = self._init_stages(
            config, 'post_batch_incremental_transforms')
        writers = self._init_stages(config, 'writers')
        tmp_dir_path = os.path.join(config['tmp_dir_path'],
                                    self.tmp_dir_suffix)
        return (reader, incremental_transforms, batch_transforms,
                post_batch_incremental_transforms, writers, tmp_dir_path)
def _init_all_stages(self, config):
        '''Create stages that are used for the pipeline.

        :param dict config: `streamcorpus_pipeline` configuration
        :return: tuple of (reader, incremental transforms, batch
          transforms, post-batch incremental transforms, writers,
          temporary directory)

        '''
        reader = self._init_stage(config, 'reader')
        incremental_transforms = self._init_stages(
            config, 'incremental_transforms')
        batch_transforms = self._init_stages(config, 'batch_transforms')
        post_batch_incremental_transforms = self._init_stages(
            config, 'post_batch_incremental_transforms')
        writers = self._init_stages(config, 'writers')
        tmp_dir_path = os.path.join(config['tmp_dir_path'],
                                    self.tmp_dir_suffix)
        return (reader, incremental_transforms, batch_transforms,
                post_batch_incremental_transforms, writers, tmp_dir_path)
Create stages that are used for the pipeline.

        :param dict config: `streamcorpus_pipeline` configuration
        :return: tuple of (reader, incremental transforms, batch
          transforms, post-batch incremental transforms, writers,
          temporary directory)