bluk load geoname.csv into ElasticSearch

es-bulk-load-geoname.py

#!/usr/bin/env python3
import sys
import time
import csv
import json
from collections import namedtuple
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

ES_ENDPOINT = 'vpc-xxxxxxxxxxxx.us-east-1.es.amazonaws.com:80'
ES_INDEX = 'geonames'
ES_TYPE = 'doc'
ES_MAPPING = {
    ES_TYPE: {
        "properties": {
            "location": {
                "type": "geo_point"
            }
        }
    }
}

def get_geoname_data(filename):
    GeonameRecord = namedtuple('GeonameRecord', 'id name ascii_name alternate_names latitude longitude feature_class feature_code country_code cc2 admin1_code admin2_code admin3_code admin4_code population elevation dem timezone modified_at')
    with open(filename, "r", encoding="utf-8") as geoname_records:
        for geoname_record in csv.reader(geoname_records):
            if len(geoname_record) == 19: # a valid row
                geoname = GeonameRecord(*geoname_record)
                yield {
                    '_index': ES_INDEX,
                    '_type': ES_TYPE,
                    '_id': geoname.id,
                    '_source': {
                        "admin1_code": geoname.admin1_code,
                        "ascii_name": geoname.ascii_name,
                        "country_code": geoname.country_code,
                        "dem": int(geoname.dem),
                        "elevation": int(geoname.elevation),
                        "feature_class": geoname.feature_class,
                        "feature_code": geoname.feature_code,
                        "id": geoname.id,
                        "location": ','.join([geoname.latitude, geoname.longitude]),
                        "modified_at": geoname.modified_at,
                        "name": geoname.name,
                        "population": int(geoname.population),
                        "timezone": geoname.timezone
                        }
                        }

if __name__ == '__main__':
    es = Elasticsearch(hosts=[ES_ENDPOINT], timeout=5000)
    es.indices.delete(index=ES_INDEX)
    es.indices.create(index=ES_INDEX)
    es.indices.put_mapping(index=ES_INDEX, doc_type=ES_TYPE, body=ES_MAPPING)
    count, _ = bulk(es, get_geoname_data('./geoname.csv'))
    print('Successfully load {0} items into ES.'.format(count))

    

geoname.csv