# Ingest data with GCP Dataflow
**URL:** https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046
**Category:** Elasticsearch
**Created:** [January 11, 2022, 2:28pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046 "2022-01-11T14:28:42Z")
**Posts on this page:** 13
**Page:** 1
### Author:  [@gustafkisi](https://discuss.elastic.co/u/gustafkisi)
#### Post date: [January 11, 2022, 2:28pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/1 "2022-01-11T14:28:42Z")
Hello.
I'm trying to ingest data to ES with the use of GCP Dataflow and the template PubSub to Elasticsearch. So far I've tried a couple of different deployments which are all on GCP using the trial option over at [elastic.co](http://elastic.co).
The data I'm trying to ingest are metrics from our devices in a simple JSON format. The dataflow is configured by using the Cloud ID for my deployment and a custom UDF to format the data from PubSub.
It all seems to go well, some metrics are ingested but then there is an error. view below:
There is always the same error. Once this error pops up no more data is ingested to ES.
Does anyone here have experience using GCP Dataflow to ingest data?
### Update
I realize the image is not the easiest to read. Here is a detailed error log from GCP:
```md
Error message from worker: java.io.IOException: Error writing to Elasticsearch, some elements could not be inserted:
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO.checkForErrors(ElasticsearchIO.java:231)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.flushBatch(ElasticsearchIO.java:1483)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.finishBundle(ElasticsearchIO.java:1449)
java.io.IOException: Error writing to Elasticsearch, some elements could not be inserted:
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO.checkForErrors(ElasticsearchIO.java:231)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.flushBatch(ElasticsearchIO.java:1483)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.finishBundle(ElasticsearchIO.java:1449)
java.io.IOException: Error writing to Elasticsearch, some elements could not be inserted:
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO.checkForErrors(ElasticsearchIO.java:231)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.flushBatch(ElasticsearchIO.java:1483)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.finishBundle(ElasticsearchIO.java:1449)
java.io.IOException: Error writing to Elasticsearch, some elements could not be inserted:
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO.checkForErrors(ElasticsearchIO.java:231)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.flushBatch(ElasticsearchIO.java:1483)
com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.finishBundle(ElasticsearchIO.java:1449)
```
This Dataflow is using default configurations so I'd really expect this to work out of the box.
---
### Author:  [@warkolm](https://discuss.elastic.co/u/warkolm)
#### Post date: [January 11, 2022, 10:56pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/2 "2022-01-11T22:56:17Z")
Welcome to our community! ๐
Please don't post pictures of text, logs or code. They are difficult to read, impossible to search and replicate (if it's code), and some people may not be even able to see them ๐
---
### Author:  [@gustafkisi](https://discuss.elastic.co/u/gustafkisi)
#### Post date: [January 12, 2022, 2:44pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/3 "2022-01-12T14:44:37Z")
Thank you! I've posted a detailed error.
---
### Author:  [@warkolm](https://discuss.elastic.co/u/warkolm)
#### Post date: [January 12, 2022, 6:58pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/4 "2022-01-12T18:58:49Z")
There's not a lot there is there โน
Is there anything in the Elasticsearch logs at that time?
---
### Author:  [@Christian\_Dahlqvist](https://discuss.elastic.co/u/Christian_Dahlqvist)
#### Post date: [January 12, 2022, 7:18pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/5 "2022-01-12T19:18:32Z")
Can you show a sample document that has been indexed and one that failed to index? It is possible that there may be a mapping conflict but without seeing JSON examples it is hard to tell. It would probably also help if you could provide the mappings for the index you are indexing into when you get the error.
---
### Author:  [@gustafkisi](https://discuss.elastic.co/u/gustafkisi)
#### Post date: [January 13, 2022, 7:55am UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/6 "2022-01-13T07:55:35Z")
> [@warkolm](#):
>
> Elasticsearch logs a
The only logs in elastic I can find is under Observability - Logs. And they show my data that I ingest. e.g.:
```json
Showing entries from Jan 11, 14:29:03
14:29:03.000
gcp.pubsub
{"points":{"udp_active":true,"udp_high":6,"udp_medium":6,"udp_low":6},"labels":{"device_id":"5000009d6bbf9cee","place_id":7800},"timestamp":"2022-01-11T13:29:03Z"}
```
This is the log I ingested to elastic, are there any other logs from elastic?
---
### Author:  [@gustafkisi](https://discuss.elastic.co/u/gustafkisi)
#### Post date: [January 13, 2022, 8:26am UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/7 "2022-01-13T08:26:44Z")
> [@Christian\_Dahlqvist](#):
>
> mappings
Found them!
Here is one of the documents, there are around 5 million similar ones:
```json
{
"_index": ".ds-logs-gcp.pubsub-metrics-2022.01.11-000001",
"_type": "_doc",
"_id": "eGtVSX4BVV1vns0ot6nA",
"_version": 1,
"_score": 1,
"_source": {
"points": {
"udp_active": true,
"udp_high": 6,
"udp_medium": 6,
"udp_low": 6
},
"labels": {
"device_id": "5000009d6bbf9cee",
"place_id": 7800
},
"@timestamp": "2022-01-11T13:29:03Z",
"agent": {
"type": "dataflow",
"name": "",
"version": "1.0.0",
"id": ""
},
"data_stream": {
"type": "logs",
"dataset": "gcp.pubsub",
"namespace": "metrics"
},
"ecs": {
"version": "1.10.0"
},
"message": "{\"points\":{\"udp_active\":true,\"udp_high\":6,\"udp_medium\":6,\"udp_low\":6},\"labels\":{\"device_id\":\"5000009d6bbf9cee\",\"place_id\":7800},\"timestamp\":\"2022-01-11T13:29:03Z\"}",
"service": {
"type": "gcp.pubsub"
},
"event": {
"module": "gcp",
"dataset": "gcp.pubsub"
}
},
"fields": {
"points.udp_medium": [
6
],
"points.udp_active": [
true
],
"points.udp_high": [
6
],
"points.udp_low": [
6
],
"data_stream.namespace": [
"metrics"
],
"labels.place_id": [
7800
],
"message": [
"{\"points\":{\"udp_active\":true,\"udp_high\":6,\"udp_medium\":6,\"udp_low\":6},\"labels\":{\"device_id\":\"5000009d6bbf9cee\",\"place_id\":7800},\"timestamp\":\"2022-01-11T13:29:03Z\"}"
],
"data_stream.type": [
"logs"
],
"service.type": [
"gcp.pubsub"
],
"agent.type": [
"dataflow"
],
"labels.device_id": [
"5000009d6bbf9cee"
],
"@timestamp": [
"2022-01-11T13:29:03.000Z"
],
"event.module": [
"gcp"
],
"agent.id": [
""
],
"ecs.version": [
"1.10.0"
],
"data_stream.dataset": [
"gcp.pubsub"
],
"agent.name": [
""
],
"agent.version": [
"1.0.0"
],
"event.dataset": [
"gcp.pubsub"
]
}
}
```
The one below it has the same index but different ids.
Edit: Actually, all 5 million documents have the same timestamp.
---
### Author:  [@Felix\_Roessel](https://discuss.elastic.co/u/Felix_Roessel)
#### Post date: [February 8, 2022, 12:48pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/8 "2022-02-08T12:48:44Z")
I was able to fix that by reducing the lifetime of the documents.
It looks like it tries to ingest the same documents again and again until the message gets dismissed.
Reducing the lifetime of the documents leads to getting data, but there is still a lag. We are in contact with google to get that fixed.
---
### Author:  [@reza1](https://discuss.elastic.co/u/reza1)
#### Post date: [February 8, 2022, 8:21pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/9 "2022-02-08T20:21:03Z")
Hello, I'm trying to follow this thread to debug my own issue with a GCP dataflow job. I have the same error message from Dataflow as shown above, but I can't find the actual logs in Observability - Logs. How can I find the log messages showing my data being ingested?
I've also tried streaming my logs and running a successful dataflow job that inserts data, but I don't see any log statements show up as a result of the job.
---
### Author:  [@warkolm](https://discuss.elastic.co/u/warkolm)
#### Post date: [February 8, 2022, 9:35pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/10 "2022-02-08T21:35:15Z")
Welcome to our community! ๐
It's better if you can please start your own topic on this, to save confusion ๐
---
### Author:  [@reza1](https://discuss.elastic.co/u/reza1)
#### Post date: [February 9, 2022, 5:09pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/11 "2022-02-09T17:09:24Z")
Hi gustafkisi. How did you locate the logs that show your ingested documents?
---
### Author:  [@reza1](https://discuss.elastic.co/u/reza1)
#### Post date: [February 9, 2022, 5:10pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/12 "2022-02-09T17:10:29Z")
Thanks warkolm. I've created a separate thread (link below).
> [@Unable to locate data ingestion logs to debug GCP Dataflow errors](https://discuss.elastic.co/t/unable-to-locate-data-ingestion-logs-to-debug-gcp-dataflow-errors/296667):
>
> Hello, I'm trying to debug an issue with a GCP Dataflow job which inserts documents into an index. I get the following errors from GCP Dataflow: java.io.IOException: Error writing to Elasticsearch, some elements could not be inserted: com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO.checkForErrors(ElasticsearchIO.java:231) com.google.cloud.teleport.v2.elasticsearch.utils.ElasticsearchIO$Write$WriteFn.flushBatch(ElasticsearchIO.java:1483) com.google.cloud.teleport.v2.elasticโฆ
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [March 9, 2022, 5:11pm UTC](https://discuss.elastic.co/t/ingest-data-with-gcp-dataflow/294046/13 "2022-03-09T17:11:09Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.