Using bulk index

Elastic Stack Elasticsearch
4

If you need to perform some processing on the docs, you may want to take a look at a couple of options:

  1. It may be that the processing you need to do can be achieved in an ingest node processor pipeline. With this approach, you define a processor pipeline and then add it as the default pipeline for the destination index, or specify the pipeline as part of the reindex operation. The .NET client documentation has an example of creating a pipeline and using it in a bulk API call.

or

  1. Use the Reindex API on the client. In contrast to ReindexOnServer, the Reindex API will pull down documents using an observable ScrollAll helper, create a destination index if need be, and bulk index documents using BulkAll as shown in previous answer. Reindex is a client specific helper function that existed before the reindex API existed in Elasticsearch (which maps to ReindexOnServer). An example Reindex would be something like
var client = new ElasticClient();
var numberOfSlices = 2; // <-- see scroll docs to see what to set this to

// set ILazyDocument type to whichever type makes sense for your data
var reindexObservable = client.Reindex<ILazyDocument>(r => r
	.BackPressureFactor(10)
	.ScrollAll("1m", numberOfSlices, s => s
		.Search(ss => ss
			.Index("amazoninf.iica.teste")
			.Source(sf => sf
				.Excludes(e => e.Fields("_d", "_n"))
			)
			.Query(q => q
				.MatchAll()
			)
		)
		.MaxDegreeOfParallelism(4) // <-- how many concurrent scroll calls
	)
	.BulkAll(b => b
		.Index("amazoninf.iica.teste.tmp")
		.Size(100) // <-- number of docs in each bulk call
		.MaxDegreeOfParallelism(2) // <-- how many concurrent bulk calls
		.RefreshOnCompleted()
	)
);

var waitHandle = new ManualResetEvent(false);
Exception exception = null;

// actions to take whilst observing the reindex process
var reindexObserver = new ReindexObserver(
	onError: e =>
	{
		exception = e;
		waitHandle.Set();
	},
	onCompleted: () => waitHandle.Set()
);

// start the observable process
reindexObservable.Subscribe(reindexObserver);

// blocking wait for reindexing process, optionally pass a timeout
waitHandle.WaitOne();

// did the observable end in exception?
if (exception != null)
	throw exception;