Title: Adding an Elastic Search index to a Dataset
Author: MotzWanted (NONE)
State: open
Created: 2021-09-09T12:21:39Z
Labels: bug

## Describe the bug
When trying to index documents from the squad dataset, the connection to ElasticSearch seems to break:

Reusing dataset squad (/Users/andreasmotz/.cache/huggingface/datasets/squad/plain_text/1.0.0/d6ec3ceb99ca480ce37cdd35555d6cb2511d223b9150cce08a837ef62ffea453)
 90%|████████████████████████████████████████████▉     | 9501/10570 [00:01<00:00, 6335.61docs/s]

No error is thrown, but the indexing breaks ~90%.

## Steps to reproduce the bug
```python
# Sample code to reproduce the bug
from datasets import load_dataset
from elasticsearch import Elasticsearch
es = Elasticsearch()
squad = load_dataset('squad', split='validation')
index_name = "corpus"
es_config = {
    "settings": {
        "number_of_shards": 1,
        "analysis": {"analyzer": {"stop_standard": {"type": "standard", " stopwords": "_english_"}}},
    },
    "mappings": {
        "properties": {
            "idx" : {"type" : "keyword"},
            "title" : {"type" : "keyword"},
            "text": {
                "type": "text",
                "analyzer": "standard",
                "similarity": "BM25"
            },
        }
    },
}
class IndexBuilder:
    """
    Elastic search indexing of a corpus
    """
    def __init__(
        self,
        *args,
        #corpus : None,
        dataset : squad,
        index_name = str,
        query = str,
        config = dict,
        **kwargs,
    ):
        #instantiate HuggingFace dataset
        self.dataset = dataset
        #instantiate ElasticSearch config
        self.config = config
        self.es = Elasticsearch()
        self.index_name = index_name
        self.query = query
    def elastic_index(self):
        print(self.es.info)
        self.es.indices.delete(index=self.index_name, ignore=[400, 404])
        search_index = self.dataset.add_elasticsearch_index(column='context', host='localhost', port='9200', es_index_name=self.index_name, es_index_config=self.config)
        return search_index
    def exact_match_method(self, index):
        scores, retrieved_examples = index.get_nearest_examples('context', query=self.query, k=1)
        return scores, retrieved_examples
if __name__ == "__main__":
    print(type(squad))
    Index = IndexBuilder(dataset=squad, index_name='corpus_index', query='Where was Chopin born?', config=es_config)
    search_index = Index.elastic_index()
    scores, examples = Index.exact_match_method(search_index)
    print(scores, examples)
    for name in squad.column_names:
        print(type(squad[name]))
```

## Environment info
We run the code in Poetry. This might be the issue, since the script runs successfully in our local environment.

Poetry:
- Python version: 3.8
- PyArrow: 4.0.1
- Elasticsearch: 7.13.4
- datasets: 1.10.2

Local:
- Python version: 3.8
- PyArrow: 3.0.0
- Elasticsearch: 7.7.1
- datasets: 1.7.0

Comments:
Hi, is this bug deterministic in your poetry env ? I mean, does it always stop at 90% or is it random ?

Also, can you try using another version of Elasticsearch ? Maybe there's an issue with the one of you poetry env

I face similar issue with oscar dataset on remote ealsticsearch instance. It was mainly due to timeout of batch indexing requests and I solve these by adding large request_timeout param in `search.py`

```
       for ok, action in es.helpers.streaming_bulk(
            client=self.es_client,
            index=index_name,
            actions=passage_generator(),
            request_timeout=3600,
        )
        ```

Hi @MotzWanted - are there any errors in the Elasticsearch cluster logs? Since it works in your local environment and the cluster versions are different between your poetry env and your local env, it is possible that it is some difference in the cluster - either settings or the cluster being under a different load etc that has this effect, so it would be useful to see if any errors are thrown in the cluster's logs when you try to ingest. 
Which elasticsearch client method is the function `add_elasticsearch_index` from your code using under the hood? Is it `helpers.bulk` or is the indexing performed using something else? You can try adding a timeout to the indexing method to see if this helps. Also, you mention that it stops at around 90%  - do you know if the timeout/hanging happens always when a particular document is being indexed or does it happen randomly at around 90% completeness but on different documents?