code: def run(self):
          
          if self.purge_existing_index:
              self.delete_index()
          self.create_index()
          es = self._init_connection()
          if self.mapping:
              es.indices.put_mapping(index=self.index, doc_type=self.doc_type,
                                     body=self.mapping)
          es.indices.put_settings({"index": {"refresh_interval": "-1"}},
                                  index=self.index)
  
          bulk(es, self._docs(), chunk_size=self.chunk_size,
               raise_on_error=self.raise_on_error)
  
          es.indices.put_settings({"index": {"refresh_interval": "1s"}},
                                  index=self.index)
          es.indices.refresh()
          self.output().touch()
docs: Run task, namely:
  
          * purge existing index, if requested (`purge_existing_index`),
          * create the index, if missing,
          * apply mappings, if given,
          * set refresh interval to -1 (disable) for performance reasons,
          * bulk index in batches of size `chunk_size` (2000),
          * set refresh interval to 1s,
          * refresh Elasticsearch,
          * create entry in marker index.