code: @Override
      public ESDeepJobConfig<T> initialize() {
          validate();
  
          configHadoop = new JobConf();
          ((JobConf) configHadoop).setInputFormat(EsInputFormat.class);
          ((JobConf) configHadoop).setOutputFormat(EsOutputFormat.class);
  
          configHadoop.set(ConfigurationOptions.ES_RESOURCE, getResource());
          configHadoop.set(ConfigurationOptions.ES_FIELD_READ_EMPTY_AS_NULL, "false");
  
          if (query != null) {
              configHadoop.set(ConfigurationOptions.ES_QUERY, query);
          }
  
          if (inputColumns != null && inputColumns.length > 0) {
              configHadoop.set(ConfigurationOptions.ES_SCROLL_FIELDS, Utils.splitListByComma(Arrays.asList
                      (inputColumns)));
          }
  
          configHadoop.set(ConfigurationOptions.ES_NODES, Utils.splitListByComma(host));
          configHadoop.set(ConfigurationOptions.ES_PORT, String.valueOf(port));
          configHadoop.set(ConfigurationOptions.ES_INPUT_JSON, "yes");
  
  
  
          if (customConfiguration != null) {
              Set<Map.Entry<String, Serializable>> set = customConfiguration.entrySet();
              Iterator<Map.Entry<String, Serializable>> iterator = set.iterator();
              while (iterator.hasNext()) {
                  Map.Entry<String, Serializable> entry = iterator.next();
                  configHadoop.set(entry.getKey(), entry.getValue().toString());
              }
  
          }
          return this;
      }
docs: {@inheritDoc}