merge_machine/abstract_data_project.py
def create_index(self, ref_path, columns_to_index, force=False, no_delete=False):
        '''Index a csv file in Elasticsearch.
        
        Unless force is set to True, this method will check if an index already
        exists with a mapping that includes that requested by columns_to_index.
        If not it will delete the existing index and fully re-index
        
        # TODO: look into re-indexing a single column
        
        Parameters
        ----------
        ref_path: str
            path to the csv file to index.
        columns_to_index: dict like {col1: list_of_analyzers1, float_col: 'float' ...}
            The analyzers (or type if not string) to use for each column.
        force: bool
            Force deleting any existing index in all cases.
        no_delete: bool
            Prevent deleting if set to True unless force is also set to True.
        '''
        
        # To solve http.client.HTTPException: got more than 100 headers
        import http
        http.client._MAXHEADERS = 1000
        
        testing = True

        dtype = {col: self._choose_dtype(col) for col in columns_to_index.keys()}
        ref_gen = pd.read_csv(ref_path, 
                          usecols=columns_to_index.keys(),
                          dtype=dtype, chunksize=self.es_insert_chunksize)
        
        if self.has_index() and (force or (not self.valid_index())):
            print('[create_index] Deleting index')
            self.ic.delete(self.index_name)
        
        columns_to_index_str = {key: val for key, val in columns_to_index.items() \
                                if not isinstance(val, str)}
        if self.has_index() and (not no_delete):
            mapping = ic.get_mapping(self.project_id)[self.project_id]['mappings']['structure']['properties']
            for col, analyzers in columns_to_index_str.items():
                if any(mapping.get(col, {'fields': None})['fields'].get(a) is None for a in analyzers):
                    print('Mapping is: {0}\nCol: {1}\nAnalyzers:{2}'.format(mapping, col, analyzers))
                    print('[create_index] Deleting index because of analyzers')
                    print('Missing:\n', [(col, a) for a in analyzers if mapping.get(col, {'fields': None})['fields'].get(a) is None])
                    logging.warning('create_index] Deleting index because of missing analyzers')
                    self.ic.delete(self.index_name)
                    break
        
        if not self.has_index():
            logging.info('Creating new index')
            log = self._init_active_log('INIT', 'transform') # TODO: is this right ?
                    
            logging.warning('Creating index')
            es_insert.create_index(self.es, self.index_name, columns_to_index,
                                   default_analyzer=DEFAULT_ANALYZER, 
                                   analyzer_definitions=ANALYZERS, force=force)
            
            
            logging.warning('Inserting in index')
            es_insert.index(es, ref_gen, self.index_name, testing, action='index')
        
            log = self._end_active_log(log, error=False)
            logging.warning('Finished indexing')
            time.sleep(5) # TODO: why is this necessary?
        else:
            logging.info('Index already exists')
        
        logging.info('Finished indexing')
        self.valid_index()
        self._write_log_buffer(written=False)
Index a csv file in Elasticsearch.
        
        Unless force is set to True, this method will check if an index already
        exists with a mapping that includes that requested by columns_to_index.
        If not it will delete the existing index and fully re-index
        
        # TODO: look into re-indexing a single column
        
        Parameters
        ----------
        ref_path: str
            path to the csv file to index.
        columns_to_index: dict like {col1: list_of_analyzers1, float_col: 'float' ...}
            The analyzers (or type if not string) to use for each column.
        force: bool
            Force deleting any existing index in all cases.
        no_delete: bool
            Prevent deleting if set to True unless force is also set to True.
        
Index a csv file in Elasticsearch.
Unless force is set to True, this method will check if an index already
exists with a mapping that includes that requested by columns_to_index.
If not it will delete the existing index and fully re-index

look into re-indexing a single column

Parameters

str
path to the csv file to index.
columns_to_index: dict like {col1: list_of_analyzers1, float_col: 'float' ...}
The analyzers (or type if not string) to use for each column.
force: bool
Force deleting any existing index in all cases.
no_delete: bool
Prevent deleting if set to True unless force is also set to True.
Index a csv file in Elasticsearch . Unless force is set to True this method will check if an index already exists with a mapping that includes that requested by columns_to_index . If not it will delete the existing index and fully re - index look into re - indexing a single column Parameters str path to the csv file to index . columns_to_index : dict like { col1 : list_of_analyzers1 float_col : ' float ' ... } The analyzers ( or type if not string ) to use for each column . force : bool Force deleting any existing index in all cases . no_delete : bool Prevent deleting if set to True unless force is also set to True .
def create_index(self, ref_path, columns_to_index, force=False, no_delete=False):
        import http
        http.client._MAXHEADERS = 1000
        testing = True
        dtype = {col: self._choose_dtype(col) for col in columns_to_index.keys()}
        ref_gen = pd.read_csv(ref_path, 
                          usecols=columns_to_index.keys(),
                          dtype=dtype, chunksize=self.es_insert_chunksize)
        if self.has_index() and (force or (not self.valid_index())):
            print('[create_index] Deleting index')
            self.ic.delete(self.index_name)
        columns_to_index_str = {key: val for key, val in columns_to_index.items() \
                                if not isinstance(val, str)}
        if self.has_index() and (not no_delete):
            mapping = ic.get_mapping(self.project_id)[self.project_id]['mappings']['structure']['properties']
            for col, analyzers in columns_to_index_str.items():
                if any(mapping.get(col, {'fields': None})['fields'].get(a) is None for a in analyzers):
                    print('Mapping is: {0}\nCol: {1}\nAnalyzers:{2}'.format(mapping, col, analyzers))
                    print('[create_index] Deleting index because of analyzers')
                    print('Missing:\n', [(col, a) for a in analyzers if mapping.get(col, {'fields': None})['fields'].get(a) is None])
                    logging.warning('create_index] Deleting index because of missing analyzers')
                    self.ic.delete(self.index_name)
                    break
        if not self.has_index():
            logging.info('Creating new index')
            log = self._init_active_log('INIT', 'transform') 
            logging.warning('Creating index')
            es_insert.create_index(self.es, self.index_name, columns_to_index,
                                   default_analyzer=DEFAULT_ANALYZER, 
                                   analyzer_definitions=ANALYZERS, force=force)
            logging.warning('Inserting in index')
            es_insert.index(es, ref_gen, self.index_name, testing, action='index')
            log = self._end_active_log(log, error=False)
            logging.warning('Finished indexing')
            time.sleep(5) 
        else:
            logging.info('Index already exists')
        logging.info('Finished indexing')
        self.valid_index()
        self._write_log_buffer(written=False)
def create_index ( self , ref_path , columns_to_index , force = False , no_delete = False ) : import http http . client . _MAXHEADERS = 1000 testing = True dtype = { col : self . _choose_dtype ( col ) for col in columns_to_index . keys ( ) } ref_gen = pd . read_csv ( ref_path , usecols = columns_to_index . keys ( ) , dtype = dtype , chunksize = self . es_insert_chunksize ) if self . has_index ( ) and ( force or ( not self . valid_index ( ) ) ) : print ( '[create_index] Deleting index' ) self . ic . delete ( self . index_name ) columns_to_index_str = { key : val for key , val in columns_to_index . items ( ) if not isinstance ( val , str ) } if self . has_index ( ) and ( not no_delete ) : mapping = ic . get_mapping ( self . project_id ) [ self . project_id ] [ 'mappings' ] [ 'structure' ] [ 'properties' ] for col , analyzers in columns_to_index_str . items ( ) : if any ( mapping . get ( col , { 'fields' : None } ) [ 'fields' ] . get ( a ) is None for a in analyzers ) : print ( 'Mapping is: {0}\nCol: {1}\nAnalyzers:{2}' . format ( mapping , col , analyzers ) ) print ( '[create_index] Deleting index because of analyzers' ) print ( 'Missing:\n' , [ ( col , a ) for a in analyzers if mapping . get ( col , { 'fields' : None } ) [ 'fields' ] . get ( a ) is None ] ) logging . warning ( 'create_index] Deleting index because of missing analyzers' ) self . ic . delete ( self . index_name ) break if not self . has_index ( ) : logging . info ( 'Creating new index' ) log = self . _init_active_log ( 'INIT' , 'transform' ) logging . warning ( 'Creating index' ) es_insert . create_index ( self . es , self . index_name , columns_to_index , default_analyzer = DEFAULT_ANALYZER , analyzer_definitions = ANALYZERS , force = force ) logging . warning ( 'Inserting in index' ) es_insert . index ( es , ref_gen , self . index_name , testing , action = 'index' ) log = self . _end_active_log ( log , error = False ) logging . warning ( 'Finished indexing' ) time . sleep ( 5 ) else : logging . info ( 'Index already exists' ) logging . info ( 'Finished indexing' ) self . valid_index ( ) self . _write_log_buffer ( written = False )
Index a csv file in Elasticsearch.
Index a csv file in Elasticsearch .
'''Index a csv file in Elasticsearch.
        
        Unless force is set to True, this method will check if an index already
        exists with a mapping that includes that requested by columns_to_index.
        If not it will delete the existing index and fully re-index
        
        # TODO: look into re-indexing a single column
        
        Parameters
        ----------
        ref_path: str
            path to the csv file to index.
        columns_to_index: dict like {col1: list_of_analyzers1, float_col: 'float' ...}
            The analyzers (or type if not string) to use for each column.
        force: bool
            Force deleting any existing index in all cases.
        no_delete: bool
            Prevent deleting if set to True unless force is also set to True.
        ''' # To solve http.client.HTTPException: got more than 100 headers # TODO: is this right ? # TODO: why is this necessary?
self
ref_path
columns_to_index
force
no_delete