code: public S initialize(ExtractorConfig extractorConfig) {
  
          setExtractorImplClassName(extractorConfig.getExtractorImplClassName());
          setEntityClass(extractorConfig.getEntityClass());
          setRddId(extractorConfig.getRddId());
          setPartitionId(extractorConfig.getPartitionId());
  
          Map<String, Serializable> values = extractorConfig.getValues();
  
          if (values.get(USERNAME) != null) {
              username(extractorConfig.getString(USERNAME));
          }
  
          if (values.get(PASSWORD) != null) {
              password(extractorConfig.getString(PASSWORD));
          }
  
          if (values.get(HOST) != null) {
              host((extractorConfig.getStringArray(HOST)));
          }
  
          if (values.get(ES_REST_PORTS) != null) {
              port((extractorConfig.getInteger(ES_REST_PORTS)));
          }
  
          if (values.get(PORT) != null) {
              port((extractorConfig.getInteger(PORT)));
          }
  
  
          if (values.get(COLLECTION) != null) {
              table(extractorConfig.getString(COLLECTION));
          }
  
          if (values.get(INPUT_COLUMNS) != null) {
              inputColumns(extractorConfig.getStringArray(INPUT_COLUMNS));
          }
  
          if (values.get(DATABASE) != null) {
              catalog(extractorConfig.getString(DATABASE));
          }
  
          if (values.get(FILTER_QUERY) != null) {
              filters(extractorConfig.getFilterArray(FILTER_QUERY));
          }
  
          return (S) this;
      }
docs: Initialize s.
  
  @param extractorConfig the extractor config
  @return the s