創(chuàng)建旅游景點圖數(shù)據(jù)庫Neo4J技術(shù)驗證

寫在前面

本章主要實踐內(nèi)容：
（1）neo4j知識圖譜庫建庫。使用導(dǎo)航poi中的公園、景點兩類csv直接建庫。
（2）pg建庫。攜程poi入庫tripdata的poibaseinfo表，之后，導(dǎo)航poi中的公園、景點也導(dǎo)入該表。

基礎(chǔ)數(shù)據(jù)建庫

python3源代碼

以下，實現(xiàn)了csv數(shù)據(jù)初始導(dǎo)入KG。如果是增量更新，代碼需要調(diào)整。
另外，星級、旅游時間 是隨機生成，不具備任何真實性。

import csv
from py2neo import *
import random
import geohashdef importCSV2NeoKG( graph,csvPath,csvType ):#單純的查詢方法node_Match = NodeMatcher(graph)seasons = ["春季","夏季","秋季","冬季"]stars = ["A","AA","AAA","AAAA","AAAAA"]with open(csvPath,"r",encoding="utf-8") as f:reader = csv.reader(f)datas = list(reader)print("csv連接成功",len(datas))newDatas = []#for data in datas:for k in range(0,len(datas)):data = datas[k]if k==0:newDatas.append(data)else:if datas[k][0]==datas[k-1][0] and datas[k][1]==datas[k-1][1]:#通過 名稱+區(qū)縣 組合判斷是否唯一continueelse:newDatas.append(data)print("去除csv中重復(fù)記錄")nodeCity_new = Node("chengshi",name="北京")cityMatch = node_Match.match("chengshi",name="北京")if cityMatch==None :graph.merge(nodeCity_new,"chengshi","name")for i in range(0,len(newDatas)):nodeQu_new = Node("quxian",name=newDatas[i][1])        rel1 = Relationship(nodeQu_new,"屬于",nodeCity_new)graph.merge(rel1,"quxian","name")geoxy_encode = geohash.encode( newDatas[i][4],newDatas[i][3],6 )nodeJingdian = Node(csvType,name=newDatas[i][0],quyu=newDatas[i][1],jianjie=newDatas[i][0],dizhi=newDatas[i][2],zuobiao=geoxy_encode)jingdianMatch = node_Match.match(csvType,name=newDatas[i][0]).where(quyu=newDatas[i][1]).first()if jingdianMatch==None :graph.create(nodeJingdian)rel2 = Relationship(nodeJingdian,"位于",nodeQu_new)graph.create(rel2)nodeTime = Node("traveltime",time=random.choice(seasons))#graph.create(nodeTime)rel3 = Relationship(nodeJingdian,"旅游時間",nodeTime)graph.merge(rel3,"traveltime","time")nodeAAA = Node("Stars",star=random.choice(stars))#graph.create(nodeAAA)rel4 = Relationship(nodeJingdian,"星級",nodeAAA)graph.merge(rel4,"Stars","star")if __name__ == '__main__':graph = Graph("bolt://localhost:7687",auth=("neo4j","neo4j?"))print("neo4j連接成功")importCSV2NeoKG(graph,"公園2050101Attr.csv","gongyuan")print("gongyuan ok")importCSV2NeoKG(graph,"景點2050201and20600102Attr.csv","jingdian")print("jingdian ok")

坐標用到了geohash，嘗試安裝過幾種geohash庫，均有錯誤。最后，直接復(fù)制源代碼生成.py文件。
geohash.py代碼如下：

from __future__ import division
from collections import namedtuple
from builtins import range
import decimal
import mathbase32 = '0123456789bcdefghjkmnpqrstuvwxyz'def _indexes(geohash):if not geohash:raise ValueError('Invalid geohash')for char in geohash:try:yield base32.index(char)except ValueError:raise ValueError('Invalid geohash')def _fixedpoint(num, bound_max, bound_min):"""Return given num with precision of 2 - log10(range)Params------num: A numberbound_max: max bound, e.g max latitude of a geohash cell(NE)bound_min: min bound, e.g min latitude of a geohash cell(SW)Returns-------A decimal"""try:decimal.getcontext().prec = math.floor(2-math.log10(bound_max- bound_min))except ValueError:decimal.getcontext().prec = 12return decimal.Decimal(num)def bounds(geohash):"""Returns SW/NE latitude/longitude bounds of a specified geohash::|      .| NE|    .  ||  .    |SW |.      |:param geohash: string, cell that bounds are required of:returns: a named tuple of namedtuples Bounds(sw(lat, lon), ne(lat, lon)). >>> bounds = geohash.bounds('ezs42')>>> bounds>>> ((42.583, -5.625), (42.627, -5.58)))>>> bounds.sw.lat>>> 42.583"""geohash = geohash.lower()even_bit = Truelat_min = -90lat_max = 90lon_min = -180lon_max = 180# 5 bits for a char. So divide the decimal by power of 2, then AND 1# to get the binary bit - fast modulo operation.for index in _indexes(geohash):for n in range(4, -1, -1):bit = (index >> n) & 1if even_bit:# longitudelon_mid = (lon_min + lon_max) / 2if bit == 1:lon_min = lon_midelse:lon_max = lon_midelse:# latitudelat_mid = (lat_min + lat_max) / 2if bit == 1:lat_min = lat_midelse:lat_max = lat_mideven_bit = not even_bitSouthWest = namedtuple('SouthWest', ['lat', 'lon'])NorthEast = namedtuple('NorthEast', ['lat', 'lon'])sw = SouthWest(lat_min, lon_min)ne = NorthEast(lat_max, lon_max)Bounds = namedtuple('Bounds', ['sw', 'ne'])return Bounds(sw, ne)def decode(geohash):"""Decode geohash to latitude/longitude. Location is approximate centre of thecell to reasonable precision.:param geohash: string, cell that bounds are required of:returns: Namedtuple with decimal lat and lon as properties.>>> geohash.decode('gkkpfve')>>> (70.2995, -27.9993)"""(lat_min, lon_min), (lat_max, lon_max) = bounds(geohash)lat = (lat_min + lat_max) / 2lon = (lon_min + lon_max) / 2lat = _fixedpoint(lat, lat_max, lat_min)lon = _fixedpoint(lon, lon_max, lon_min)Point = namedtuple('Point', ['lat', 'lon'])return Point(lat, lon)def encode(lat, lon, precision):"""Encode latitude, longitude to a geohash.:param lat: latitude, a number or string that can be converted to decimal.Ideally pass a string to avoid floating point uncertainties.It will be converted to decimal.:param lon: longitude, a number or string that can be converted to decimal.Ideally pass a string to avoid floating point uncertainties.It will be converted to decimal.:param precision: integer, 1 to 12 represeting geohash levels upto 12.:returns: geohash as string.>>> geohash.encode('70.2995', '-27.9993', 7)>>> gkkpfve"""lat = decimal.Decimal(lat)lon = decimal.Decimal(lon)index = 0  # index into base32 mapbit = 0   # each char holds 5 bitseven_bit = Truelat_min = -90lat_max = 90lon_min = -180lon_max = 180ghash = []while(len(ghash) < precision):if even_bit:# bisect E-W longitudelon_mid = (lon_min + lon_max) / 2if lon >= lon_mid:index = index * 2 + 1lon_min = lon_midelse:index = index * 2lon_max = lon_midelse:# bisect N-S latitudelat_mid = (lat_min + lat_max) / 2if lat >= lat_mid:index = index * 2 + 1lat_min = lat_midelse:index = index * 2lat_max = lat_mideven_bit = not even_bitbit += 1if bit == 5:# 5 bits gives a char in geohash. Start overghash.append(base32[index])bit = 0index = 0return ''.join(ghash)def adjacent(geohash, direction):"""Determines adjacent cell in given direction.:param geohash: cell to which adjacent cell is required:param direction: direction from geohash, string, one of n, s, e, w:returns: geohash of adjacent cell>>> geohash.adjacent('gcpuyph', 'n')>>> gcpuypk"""if not geohash:raise ValueError('Invalid geohash')if direction not in ('nsew'):raise ValueError('Invalid direction')neighbour = {'n': ['p0r21436x8zb9dcf5h7kjnmqesgutwvy','bc01fg45238967deuvhjyznpkmstqrwx'],'s': ['14365h7k9dcfesgujnmqp0r2twvyx8zb','238967debc01fg45kmstqrwxuvhjyznp'],'e': ['bc01fg45238967deuvhjyznpkmstqrwx','p0r21436x8zb9dcf5h7kjnmqesgutwvy'],'w': ['238967debc01fg45kmstqrwxuvhjyznp','14365h7k9dcfesgujnmqp0r2twvyx8zb'],}border = {'n': ['prxz',     'bcfguvyz'],'s': ['028b',     '0145hjnp'],'e': ['bcfguvyz', 'prxz'],'w': ['0145hjnp', '028b'],}last_char = geohash[-1]parent = geohash[:-1]  # parent is hash without last chartyp = len(geohash) % 2# check for edge-cases which don't share common prefixif last_char in border[direction][typ] and parent:parent = adjacent(parent, direction)index = neighbour[direction][typ].index(last_char)return parent + base32[index]def neighbours(geohash):"""Returns all 8 adjacent cells to specified geohash::| nw | n | ne ||  w | * | e  || sw | s | se |:param geohash: string, geohash neighbours are required of:returns: neighbours as namedtuple of geohashes with properties n,ne,e,se,s,sw,w,nw>>> neighbours = geohash.neighbours('gcpuyph')>>> neighbours>>> ('gcpuypk', 'gcpuypm', 'gcpuypj', 'gcpuynv', 'gcpuynu', 'gcpuyng', 'gcpuyp5', 'gcpuyp7')>>> neighbours.ne>>> gcpuypm"""n = adjacent(geohash, 'n')ne = adjacent(n, 'e')e = adjacent(geohash, 'e')s = adjacent(geohash, 's')se = adjacent(s, 'e')w = adjacent(geohash, 'w')sw = adjacent(s, 'w')nw = adjacent(n, 'w')Neighbours = namedtuple('Neighbours',['n', 'ne', 'e', 'se', 's', 'sw', 'w', 'nw'])return Neighbours(n, ne, e, se, s, sw, w, nw)

KG效果

命令行里啟動neo4j：
neo4j.bat console

KG入庫效率優(yōu)化方案

上文的python方法是py2neo的基本方法，經(jīng)過本人親測，當節(jié)點量到3~5w的時候，入庫開始變慢，以小時計。

百度后，有大神提供了另外一種方法：
采用這種方法，建立50w個節(jié)點和50w個關(guān)系，流程包括node、rel的建立、append到list、入庫，全過程4分鐘以內(nèi)搞定。測試環(huán)境在VM虛擬機實現(xiàn)。
代碼如下：

from py2neo import Graph, Subgraph, Node, Relationship
from progressbar import *
import datetimedef batch_create(graph, nodes_list, relations_list):subgraph = Subgraph(nodes_list, relations_list)tx_ = graph.begin()tx_.create(subgraph)graph.commit(tx_)if __name__ == '__main__':# 連接neo4jgraph = Graph("bolt://localhost:7687",auth=("neo4j","neo4j?"))# 批量創(chuàng)建節(jié)點nodes_list = []  # 一批節(jié)點數(shù)據(jù)relations_list = []  # 一批關(guān)系數(shù)據(jù)nodeCity_new = Node("chengshi",name="北京")nodes_list.append(nodeCity_new)widgets = ['CSV導(dǎo)入KG進度: ', Percentage(), ' ', Bar('#'), ' ', Timer(), ' ', ETA(), ' ']bar = ProgressBar(widgets=widgets, maxval=500000)bar.start()#for i in range(0,500000):bar.update(i+1)nodeQu_new = Node("quxian",name="Test{0}".format(i))nodes_list.append(nodeQu_new)rel1 = Relationship(nodeQu_new,"屬于",nodeCity_new)relations_list.append(rel1)bar.finish()current_time = datetime.datetime.now()print("current_time:    " + str(current_time))# 批量創(chuàng)建節(jié)點/關(guān)系batch_create(graph, nodes_list, relations_list)current_time = datetime.datetime.now()print("current_time:    " + str(current_time))print("batch ok")

PostGreSQL建庫

pg建庫。攜程poi入庫tripdata的poibaseinfo表，之后，導(dǎo)航poi中的公園、景點也導(dǎo)入該表。

攜程poi導(dǎo)入代碼：psycopg2_004.py

import psycopg2
import csv
import random
import geohash
from progressbar import *#
#攜程爬蟲csv數(shù)據(jù)入庫
# 
def importCtripCSV2PG(cur,csvpath,csvcity,csvprovice):#     csvPath = "pois_bj_ctrip.csv"with open(csvpath,"r",encoding="utf-8") as f:reader = csv.reader(f)datas = list(reader)print("csv datas number = {}".format(len(datas)))print("")widgets = ['爬蟲數(shù)據(jù)導(dǎo)入PG進度: ', Percentage(), ' ', Bar('#'), ' ', Timer(), ' ', ETA(), ' ']bar = ProgressBar(widgets=widgets, maxval=len(datas))bar.start()##sCol = "namec,namec2,namee,tags,brief,ticket,ticketmin,ticketadult,ticketchild,ticketold,ticketstudent,scores,scorenumber,opentime,spendtime,introduceinfo,salesinfo,guid,quyu,city,province,contry"#sCol = "namec,namee,tags,brief,ticket,ticketmin,ticketadult,ticketchild,ticketold,ticketstudent,scores,scorenumber,opentime,spendtime,introduceinfo,salesinfo,city,province,contry"sCol = "namec,namee,tags,brief,ticket,ticketmin,ticketadult,ticketchild,ticketold,ticketstudent,scores,scorenumber,opentime,spendtime,introduceinfo,salesinfo,x,y,geos,photourl,city,province,contry"#     print("sCol number = {}".format(len(sCol)))for i in range(0,len(datas)):bar.update(i+1)data = datas[i]if data==None or len(data)==0:#print("{}行None值".format(i))continue if data[0]=="名稱" or data[0]==None :continue geoxy_encode = geohash.encode( data[5],data[4],7 )values = ",".join("\'{0}\'".format(w) for w in [data[0].replace("\'","''"),data[1].replace("\'","''"),data[6],data[7],data[8],data[9],data[13],data[16],data[14],data[15],data[10],data[11],data[18].replace("\'","''"),data[17].replace("\'","''"),data[19].replace("\'","''"),data[20].replace("\'","''"),data[5],data[4],geoxy_encode,data[12],csvcity,csvprovice,"中國"])#     print(values)sqlpre = "insert into poibaseinfo({})".format(sCol)sql = sqlpre+" values ({})".format(values)#     print(sql)try:cur.execute(sql)except psycopg2.Error as e:print(e)bar.finish()if __name__ == '__main__':user = "postgres"pwd = "你的密碼"port = "5432"hostname = "127.0.0.1"conn = psycopg2.connect(database = "tripdata", user = user, password = pwd, host = "127.0.0.1", port = port)print(conn)sql = "select * from poibaseinfo"cur = conn.cursor()cur.execute(sql)cols = cur.descriptionprint("PG cols number = {}".format(len(cols)))#CSV文件導(dǎo)入PGcsvPath = "pois_bj_ctrip.csv"    importCtripCSV2PG(cur,csvPath,"北京","北京")#其他CSV文件導(dǎo)入PG#TODO...conn.commit()cur.close()conn.close()print("ok")