# Ingestion not consistent, data loss

**URL:** <https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512>\
**Category:** Logstash\
**Created:** [April 21, 2023, 10:59am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512 "2023-04-21T10:59:24Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![Rui\_Goncalves](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rui_goncalves/32/72197_2.png) [@Rui\_Goncalves](https://discuss.elastic.co/u/Rui_Goncalves)\
**Post date:** [April 21, 2023, 10:59am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/1 "2023-04-21T10:59:25Z")

</div>

Hi,  
We're ingesting a big amount of data from an oracle database into elastic through logstash.  
We import some oracle table information into different arrays in elastic. Each table corresponding to an array, and each row corresponding to an element of the array.  
Sometimes it imports lets say 5 rows from the table information and shows a 5 element array.  
Other times it imports those same 5 rows but just shows 2 or 3 array elements in elastic.  
This it's a random behavior and it happens with arrays holding different types of information.

Using the stdout plugin I can see that the indeed 5 elements of the array reach the output stage on logstash.  
What can be happening so that then they don't all appear on elastic sometimes.

Could it be that they are not sent to elastic at all, or elastic it's not able to index it as it's a big load of information being processed.  
I don't see any issue reported on logstash logs.  
Do you have an idea on what can be the issue or even if not  
how can I check what is reaching elastic and what can I tune on the output plugin or on logstash/elastic config files?  
Any hint would be much appreciated as we're running out of options.  
Thanks

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [April 23, 2023, 11:51pm UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/2 "2023-04-23T23:51:52Z")

</div>

Hello,

Please share your logstash configuration and a sample of your message.

> [@Rui\_Goncalves](#):
>
> This it's a random behavior and it happens with arrays holding different types of information.

Depending on how the data looks like and the mapping of the issues, this could lead to some documents not being indexed, but if Elasticsearch has some problem indexing a document you would get a log line in Logstash log.

You have nothing in Logstash logs?

Can you share an example of how those different types looks like?

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [April 24, 2023, 2:42am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/3 "2023-04-24T02:42:48Z")

</div>

Are you using a jdbc filter? That will return each row as a separate event. If you are combining those events into an event that has an array of rows then you will need to run with --pipeline.workers 1, otherwise you could have the rows spread across multiple worker threads.

---

<div class="post-metadata">

**Author:** ![Rui\_Goncalves](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rui_goncalves/32/72197_2.png) [@Rui\_Goncalves](https://discuss.elastic.co/u/Rui_Goncalves)\
**Post date:** [April 24, 2023, 9:11am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/5 "2023-04-24T09:11:33Z")

</div>

Thanks Leandro, I've shared the config file and ruby file.

---

<div class="post-metadata">

**Author:** ![Rui\_Goncalves](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rui_goncalves/32/72197_2.png) [@Rui\_Goncalves](https://discuss.elastic.co/u/Rui_Goncalves)\
**Post date:** [April 24, 2023, 9:13am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/6 "2023-04-24T09:13:26Z")

</div>

Thanks Badger. We've pipeline.workers: 5. I've thought about something of that sort, and I've tested with just 1 worker, it get's slow as expected but the main issue is that I was having read timeout on the input plugin with that. Do you have any suggestion to avoid that timeout with just 1 worker?  
Here's a sample of a config file that we're using

```auto
input {
  jdbc {
    lowercase_column_names => true
    jdbc_driver_library => ""
    jdbc_driver_class => "Java::oracle.jdbc.driver.OracleDriver"
	jdbc_connection_string => "jdbc:oracle:thin:@${DB_HOST_SYSPER}:${DB_PORT_SYSPER}/${DB_SERVICE_SYSPER}"
	jdbc_user => "${DB_USER_SYSPER}"
	jdbc_password => "${DB_PASS_SYSPER}"
    tracking_column => "per_id"
    statement => "..."
  }
}

filter {
    aggregate {
        task_id => "%{per_id}"
        code => "
        map['per_id'] = event.get('per_id')
        map['jd_knowledges'] ||= []
        map['jd_knowledges'] << event.to_hash # JSON.parse(event.to_json) # a small hack to get event data
        event.cancel()
        "
        push_previous_map_as_event => true
        timeout => 30
    }
	
	fingerprint {
	  key => "HR_SEARCH_KEY"
	  source => "per_id"
	  base64encode => true
	  target => "[@metadata][fingerprint]"
	  method => "SHA1"
	}

    elasticsearch {
        hosts => ["${ELASTICSEARCH_NODE_1}", "${ELASTICSEARCH_NODE_2}", "${ELASTICSEARCH_NODE_3}"]
        user => "${ELASTIC_USER}"
        password => "${ELASTIC_PASSWORD}"
        ca_file => "${LOGSTASH_CERTS}/cacert.pem"
        index => "${INDEX_NAME}"
        query => '_id:"person:%{[@metadata][fingerprint]}"'
        fields => { "[car][current_assignment][job_description]" => "job_description" }
    }

    ruby {
        path => "${LOGSTASH_DATA_PATH}/task5.rb"
    }
	
	mutate {
        remove_field => ["per_id", "jd_knowledges", "job_description"]
	}

    sleep {
        time => "1" # Sleep 1 second
        every => 250 # on every 250th event
    }
}

output {
    elasticsearch {
        hosts => ["${ELASTICSEARCH_NODE_1}", "${ELASTICSEARCH_NODE_2}", "${ELASTICSEARCH_NODE_3}"]
        user => "${ELASTIC_USER}"
        password => "${ELASTIC_PASSWORD}"
        ssl => true
        cacert => "${LOGSTASH_CERTS}/cacert.pem"
        index => "${INDEX_NAME}"
        document_id => "person:%{[@metadata][fingerprint]}" #"person:%{perId}"
        doc_as_upsert => true
        action => "update"
        http_compression => true
    }
}

```

And the ruby file used

```auto
def filter(event)
	knowledges_data = event.get('jd_knowledges') ? event.get('jd_knowledges') : []

  if !(knowledges_data.nil? || knowledges_data.empty?)

		job_description = event.get('job_description') ? event.get('job_description') : {}

		knowledges_hash = {}
		
		knowledges_data.each do |function_entry|
			key = function_entry['jik_id'].to_s
			knowledges_hash[key] ||= {}
			
			%w( jdl_id jdl_descr jdl_obsolete jdl_id_reports_to jdl_descr_reports_to level_id jdl_id_parent_lev_2 jdl_id_parent_descr_lev_2 jdl_id_parent_lev_1 jdl_id_parent_descr_lev_1 total_cnt partial_cnt ).each do |field_name|
			  value = function_entry[field_name]
			  knowledges_hash[key][:"#{field_name}"] ||= value if value 
			end
			
		end 
		
		jd_knowledges = []
		knowledges_hash.each do |jik_id, params_obj|
			if jik_id
			  params_obj[:jik_id] = jik_id.to_i
			  jd_knowledges.push(params_obj)
			end
		end
		
		job_description['knowledges'] = jd_knowledges
		event.set('[car][current_assignment][job_description]', job_description)
		[event]
	end
	    
end 

```

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [April 24, 2023, 11:56am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/7 "2023-04-24T11:56:09Z")

</div>

> [@Rui\_Goncalves](#):
>
> We've pipeline.workers: 5. I've thought about something of that sort, and I've tested with just 1 worker, it get's slow as expected but the main issue is that I was having read timeout on the input plugin with that.

As @Badger mentioned, you need to use `pipeline.workers` set to `1` if you are using the `aggregate` filter, it is required for the `aggregate` filter to work and yes, one of the drawbacks is that it will make processing slow as you are using only one core.

Can you share a log of this timeout? The number of workers should have some impact on the `filter` and `output` block only, but not in the `input` block.

---

<div class="post-metadata">

**Author:** ![Rui\_Goncalves](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/rui_goncalves/32/72197_2.png) [@Rui\_Goncalves](https://discuss.elastic.co/u/Rui_Goncalves)\
**Post date:** [May 4, 2023, 6:44am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/8 "2023-05-04T06:44:06Z")

</div>

Thanks @leandrojmp and @Badger for your great help. Setting the nbr of workers as 1 solved the issue. There was no timeout in the subsequent tests.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [June 1, 2023, 6:44am UTC](https://discuss.elastic.co/t/ingestion-not-consistent-data-loss/330512/9 "2023-06-01T06:44:17Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.