Copies an ElasticSearch index between two servers using Bulk method

es_index_copier

from __future__ import unicode_literals
import requests
import json
from io import StringIO
from requests.auth import HTTPBasicAuth
import datetime, pytz

def copy_indexes(sourceSearchUrl, destinationBulkUrl, newIndexName=None, newTypeName=None, auth=None):

  BatchSize = 40
  if sourceSearchUrl.find("?") < 0:
      sourceSearchUrl += "?"

  for i in range(0, 100000):
    reqUrl = sourceSearchUrl + "from=" + str(i*BatchSize) + "&size=" + str(BatchSize)
    print(reqUrl)
    result = requests.get(reqUrl, auth=auth, verify=False)
    data = json.loads(result.text)
    entries = data["hits"]["hits"]

    count = len(entries)
    if count == 0:
      break

    esb = ESDataBuilder()
    for j in range(0, count):
      record = entries[j]
      # index
      indexName = newIndexName
      if indexName is None:
        indexName = record["_index"]

      # type
      typeName = newTypeName
      if typeName is None:
        typeName = record["_type"]


      esb.add_operation(record["_source"], indexName, typeName, record["_id"])

    copy_bulk(esb, destinationBulkUrl)
    print(BatchSize*i)




def copy_index(data, destUrl):
  requests.put(destUrl + data["_id"],  json.dumps(data["_source"]))

def copy_bulk(dataBuilder, destUrl):
  requests.put(destUrl, dataBuilder.toString())



class ESDataBuilder:

  def __init__(self):
    self.buffer = StringIO()

  def add_operation(self, data, indexName, typeName, theId):
    self.buffer.write(json.dumps({"index": {"_index": indexName, "_type": typeName, "_id": theId}}))
    self.buffer.write("\n")
    self.buffer.write(json.dumps(data))
    self.buffer.write("\n")

  def toString(self):
    return self.buffer.getvalue()