How to ensure all data is captured in ES API?

user_9972301 0
Viewed 96

I am trying to create an API in Python to pull the data from ES and feed it in a data warehouse. The data is live and being filled every second so I am going to create a near-real-time pipeline.

The current URL format is {{url}}/{{index}}/_search and the test payload I am sending is:

{
   "from" : 0,
   "size" : 5
}

On the next refresh it will pull using payload:

{
   "from" : 6,
   "size" : 5
}

And so on until it reaches the total amount of records. The PROD environment has about 250M rows and I'll set the size to 10 K per extract.

I am worried though as I don't know if the records are being reordered within ES. Currently, there is a plugin which uses a timestamp generated by the user but that is flawed as sometimes documents are being skipped due to a delay in the jsons being made available for extract in ES and possibly the way the time is being generated.

Does anyone know what is the default sorting when pulling the data using /_search?

elasticsearch python
2 Answers

I suppose what you're looking for is a streaming / changes API which is nicely described by @Val here and also an open feature request.

I the meantime, you cannot really count on the size and from parameters -- you could probably make redundant queries and handle the duplicates before they reach your data warehouse.

Another option would be to skip ES in this regard and stream directly to the warehouse? What I mean is, take an ES snapshot up until a given time once (so you keep the historical data), feed it to the warehouse, and then stream directly from where ever you're getting your data to the warehouse.

Addendum

AFAIK the default sorting is by the insertion date. But there's no internal _insertTime or similar.. You can use cursors -- it's called scrolling and here's a py implementation. But this goes from the 'latest' doc to the 'first', not vice versa. So it'll give you all the existing docs but I'm not so sure about the newly added ones while you were scrolling. You'd then wanna run the scroll again which is suboptimal.

You could also pre-sort your index which should work quite nicely for your use case when combined w/ scrolling.

user_81603180

Thanks for the responses. After consideration with my colleagues, we decided to implement and use the _ingest API instead to create a pipeline in ES which inserts the server document ingestion date on each doc.

Steps:

  1. Create the timestamp pipeline
PUT _ingest/pipeline/timestamp_pipeline
{
  "description" : "Inserts timestamp field for all documents",
  "processors" : [
    {
      "set" : {
        "field": "insert_date",
        "value": "{{_ingest.timestamp}}"
      }
    }
  ]
}
  1. Update indexes to add the new default field
PUT /*/_settings
{
  "index" : {
    "default_pipeline": "timestamp_pipeline"
  }
}
  1. In Python then I use the _scroll API like so:
    es = Elasticsearch(cfg.esUrl, port = cfg.esPort, timeout = 200)
    doc = {
      "query": {
        "range": {
          "insert_date": {
            "gte": lastRowDateOffset
          }
        }
      }
    }

    res = es.search(
        index = Index,
        sort = "insert_date:asc",
        scroll = "2m",
        size = NumberOfResultsPerPage,
        body = doc
    )

Where lastRowDateOffset is the date of the last run

user_99723010
Related