将数据导入MongoDB集群

Posted 2020-12-03 ParamousGIS

import sys
import json
import pymongo
import datetime
from pymongo import MongoClient

client = MongoClient(‘mongodb://192.168.1.31:20000,192.168.1.34:20000‘)
db = client.RHY
collection = db.ST_RIVER_R

f = open("D:/bigdata/st_river_r.CSV")
line = f.readline()
print(line)
fieldNames = line.split(‘,‘)
# STCD,TM,Z,Q,XSA,XSAVV,XSMXV,FLWCHRCD,WPTN,MSQMT,MSAMT,MSVMT
line = f.readline()
count = 0
records = []
insertCount = 0
while line:
     #
     count = count + 1
     fieldValues = line.split(‘,‘)
     if len(fieldValues) == 12 or fieldValues[0].strip() != ‘‘:
         insertObj = {}
         STCD = fieldValues[0]
         insertObj[‘STCD‘] = STCD
         TM = fieldValues[1]
         if TM.strip() != ‘‘:
             TM = datetime.datetime.strptime(TM, ‘%Y-%m-%d %H:%M:%S‘)
             insertObj[‘TM‘] = TM
         Z = fieldValues[2]
         if Z.strip() != ‘‘:
             Z = float(Z)
             insertObj[‘Z‘] = Z
         Q = fieldValues[3]
         if Q.strip() != ‘‘:
             Q = float(Q)
             insertObj[‘Q‘] = Q
         # XSA
         XSA = fieldValues[4]
         if XSA.strip() != ‘‘:
             XSA = float(XSA)
             insertObj[‘XSA‘] = XSA
         # XSAVV
         XSAVV = fieldValues[5]
         if XSAVV.strip() != ‘‘:
             XSAVV = float(XSAVV)
             insertObj[‘XSAVV‘] = XSAVV
         #
         XSMXV = fieldValues[6]
         if XSMXV.strip() != ‘‘:
             XSMXV = float(XSMXV)
             insertObj[‘XSMXV‘] = XSMXV
         #
         FLWCHRCD = fieldValues[7]
         if FLWCHRCD.strip() != ‘‘:
             insertObj[‘FLWCHRCD‘] = FLWCHRCD
         #
         WPTN = fieldValues[8]
         if WPTN.strip() != ‘‘:
             insertObj[‘WPTN‘] = WPTN
         #
         MSQMT = fieldValues[9]
         if MSQMT.strip() != ‘‘:
             insertObj[‘MSQMT‘] = MSQMT
         #
         MSAMT = fieldValues[10]
         if MSAMT.strip() != ‘‘:
             insertObj[‘MSAMT‘] = MSAMT
         #
         MSVMT = fieldValues[11]
         if MSVMT.strip() != ‘‘:
             insertObj[‘MSVMT‘] = MSVMT
         #
         # collection.insert_one(insertObj)
         # collection.insert_many(new_posts)
         records.append(insertObj)
         if len(records) == 1000:
             insertCount = insertCount + 1
             if count > 1451000:
                 collection.insert_many(records)
                 print(str(count) + ‘  ‘ + str(insertCount))
             print(count)
             records = []
     else:
         print(line)
     #
     line = f.readline()

f.close()
client.close()