Script to load and geocode MOT data

LoadMOTs

import gzip
import csv
from elasticsearch import helpers
from elasticsearch.client import Elasticsearch
import time

csv.register_dialect('piper', delimiter='|', quoting=csv.QUOTE_NONE)

# See http://postcodepal.com/dbgen/postcode_areas_true-centroids.zip 
pf=open('/Users/Mark/Documents/work/irdata/MOT/postcode_areas.csv')
preader = csv.reader(pf)
centroids={};
for prow in preader:
    print prow
    pcArea=prow[0]
    pcTown=prow[4]
    pcLat=prow[12]
    pcLon=prow[13]
    centroids[pcArea]={
        "town":pcTown,
        "lat":pcLat,
        "lon":pcLon
    }
pf.close()
# See http://data.gov.uk/dataset/anonymised_mot_test and test results
f=gzip.open('/Users/Mark/Documents/work/irdata/MOT/test_result_2013.txt.gz')

start = time.time()
es = Elasticsearch()



actions = []

reader = csv.reader(f, delimiter='|')
lc=0
for row in reader:
    doc={}
    doc["ID"]=row[0]
    doc["VehicleID"]=row[1]
    doc["TestDate"]=row[2]
    doc["TestClassID"]=row[3]
    doc["TestType"]=row[4]
    doc["TestResult"]=row[5]
    doc["TestMileage"]=row[6]
    doc["PostcodeArea"]=row[7]
    doc["Make"]=row[8]
    doc["Model"]=row[9]
    doc["Colour"]=row[10]
    doc["FuelType"]=row[11]
    doc["CylinderCapacity"]=row[12]
    doc["FirstUseDate"]=row[13]

    centroid=centroids[doc["PostcodeArea"]]
    if centroid is not None:
        #print centroid
        doc["town"]=centroid["town"],
        doc["location"]={
            "lat":centroid["lat"],
            "lon":centroid["lon"]
                    }
    else:
        print doc["PostcodeArea"]
    action = {
                        "_index": "mots",
                        '_op_type': 'index',
                        "_type": "mot",
                        "_source": doc
                    }

    actions.append(action)
    # Flush bulk indexing action if necessary
    if len(actions) >= 5000:
        helpers.bulk(es, actions)
        ## TO check for failures and take appropriate action
        del actions[0:len(actions)]


# Flush
if len(actions) >0:
    helpers.bulk(es, actions)
    del actions[0:len(actions)]

f.close()

Mapping

{
   "settings" : {
        "number_of_shards" : 1,
        "number_of_replicas" : 0
    },
	"mappings":{
		"mot": {
			"dynamic_templates" : [ {
					  "string_fields" : {
						"mapping" : {
						  "index" : "analyzed",
						  "omit_norms" : true,
						  "type" : "string",
						  "fields" : {
							"raw" : {
							  "index" : "not_analyzed",
							  "ignore_above" : 256,
							  "type" : "string"
							}
						  }
						},
						"match" : "*",
						"match_mapping_type" : "string"
					  }
					} ],		
		
            "properties": {
               	"TestDate" : {
		            "type" : "date",
		            "format" : "YYYY-MM-dd"
	        },
		"FirstUseDate" : {
		            "type" : "date",
		            "format" : "YYYY-MM-dd"
                },
              "location" : {
                "type" : "geo_point"
              }                
            }
        }        
    }        
}