@Override
protected void setUp() throws Exception {
    // create ES instance
    runner = new ElasticsearchClusterRunner();

    // create ES nodes
    runner.onBuild(new ElasticsearchClusterRunner.Builder() {
        @Override
        public void build(final int number, final Builder settingBuilder) {
        }
    }).build(
            newConfigs()
                    .clusterName("es-auth" + System.currentTimeMillis())
                    .ramIndexStore().numOfNode(1));

    // wait for yellow status
    runner.ensureYellow();
}
 

@Override
public void open(JobConf job, String name) throws IOException {
  clusterName = job.get(ElasticConstants.CLUSTER);
  host = job.get(ElasticConstants.HOST);
  port = job.getInt(ElasticConstants.PORT, -1);

  Builder settingsBuilder = ImmutableSettings.settingsBuilder();
  
  BufferedReader reader = new BufferedReader(job.getConfResourceAsReader("elasticsearch.conf"));
  String line;
  String parts[];

  while ((line = reader.readLine()) != null) {
    if (StringUtils.isNotBlank(line) && !line.startsWith("#")) {
      line.trim();
      parts = line.split("=");

      if (parts.length == 2) {
        settingsBuilder.put(parts[0].trim(), parts[1].trim());
      }
    }
  }

  // Set the cluster name and build the settings
  Settings settings = settingsBuilder.put("cluster.name", clusterName).build();
  
  // Prefer TransportClient
  if (host != null && port > 1) {
    client = new TransportClient(settings).addTransportAddress(new InetSocketTransportAddress(host, port));
  } else if (clusterName != null) {
    node = nodeBuilder().settings(settings).client(true).node();
    client = node.client();
  }

  bulk = client.prepareBulk();
  defaultIndex = job.get(ElasticConstants.INDEX, "nutch");
  maxBulkDocs = job.getInt(
          ElasticConstants.MAX_BULK_DOCS, DEFAULT_MAX_BULK_DOCS);
  maxBulkLength = job.getInt(
          ElasticConstants.MAX_BULK_LENGTH, DEFAULT_MAX_BULK_LENGTH);
}