Add pipeline to bulk params

elastic · Mar 4, 2020 · 63c24c9 · 63c24c9
1 parent 1c34f6c
commit 63c24c9
Showing 1 changed file with 10 additions and 4 deletions.
diff --git a/elasticsearch-model/lib/elasticsearch/model/importing.rb b/elasticsearch-model/lib/elasticsearch/model/importing.rb
@@ -143,6 +143,7 @@ def import(options={}, &block)
           target_index = options.delete(:index)     || index_name
           target_type  = options.delete(:type)      || document_type
           transform    = options.delete(:transform) || __transform
+          pipeline     = options.delete(:pipeline)
           return_value = options.delete(:return)    || 'count'
 
           unless transform.respond_to?(:call)
@@ -158,10 +159,15 @@ def import(options={}, &block)
           end
 
           __find_in_batches(options) do |batch|
-            response = client.bulk \
-                         index:   target_index,
-                         type:    target_type,
-                         body:    __batch_to_bulk(batch, transform)
+            params = {
+              index: target_index,
+              type:  target_type,
+              body:  __batch_to_bulk(batch, transform)
+            }
+
+            params[:pipeline] = pipeline if pipeline
+
+            response = client.bulk params
 
             yield response if block_given?