Added a simple reindex helper
This commit is contained in:
@@ -41,3 +41,13 @@ def scan(client, query=None, scroll='5m', **kwargs):
|
||||
yield hit
|
||||
scroll_id = resp['_scroll_id']
|
||||
|
||||
def reindex(client, source_index, target_index, target_client=None, chunk_size=500, scroll='5m'):
|
||||
target_client = client if target_client is None else target_index
|
||||
|
||||
docs = scan(client, index=source_index, scroll=scroll)
|
||||
def _change_doc_index(hits, index):
|
||||
for h in hits:
|
||||
h['_index'] = index
|
||||
yield h
|
||||
|
||||
return bulk_index(target_client, _change_doc_index(docs, target_index), chunk_size=chunk_size)
|
||||
|
||||
@@ -24,3 +24,20 @@ class TestScan(ElasticTestCase):
|
||||
self.assertEquals(100, len(docs))
|
||||
self.assertEquals(set(map(str, range(100))), set(d['_id'] for d in docs))
|
||||
self.assertEquals(set(range(100)), set(d['_source']['answer'] for d in docs))
|
||||
|
||||
class TestReindex(ElasticTestCase):
|
||||
def test_all_documents_get_moved(self):
|
||||
bulk = []
|
||||
for x in range(100):
|
||||
bulk.append({"index": {"_index": "test_index", "_type": "answers" if x % 2 == 0 else "questions", "_id": x}})
|
||||
bulk.append({"answer": x, "correct": x == 42})
|
||||
self.client.bulk(bulk, refresh=True)
|
||||
|
||||
helpers.reindex(self.client, "test_index", "prod_index")
|
||||
self.client.indices.refresh()
|
||||
|
||||
self.assertTrue(self.client.indices.exists("prod_index"))
|
||||
self.assertTrue(self.client.indices.exists_type("prod_index", "answers"))
|
||||
self.assertTrue(self.client.indices.exists_type("prod_index", "questions"))
|
||||
self.assertEquals(100, self.client.count(index='prod_index')['count'])
|
||||
|
||||
|
||||
Reference in New Issue
Block a user