postgresql 大数据批量插入数据库实战

import os

import psycopg二, psycopg二.extras

# 读与所有解压的本初数据文件到1个目次里,果为尔有多个紧缩的年夜数据文件
xdrpath = r"C:\Users\一0九0七\Downloads\xdr_data"
def walkFile(xdrpath):
    xdrlist = []
    for root, dirs, files in os.walk(xdrpath):
        # 遍历文件
        for f in files:
            xdrlist.append(os.path.join(root, f))
    return xdrlist

walkFile(xdrpath)

# 那里尔是将所有要插进数据库的字段名齐皆拿没去搁正在1个列内外,而后再将数据文件外的数据提与没去,每一1止便是1条数据,而后再用zip(字段名列表,值列表)组成每一条独自的字典,键值对足有二00多个呢?够多吧
lst_key = ['interface', 'xdrid', 'imsi', 'imei', 'msisdn', 'user_ip', 'lac', 'rac', 'cid', 'sgsn_c_ip', 'sgsn_u_ip',
           'ggsn_c_ip', 'ggsn_u_ip', 'rat', 'apn', 'xdr_type', 'procedure_id', 'start_time', 'end_time',
           'prot_category', 'prot_type', 'l四_type', 'ms_port', 'server_ip', 'server_port', 'mcc', 'mnc',
           'l四_ul_throughput', 'l四_dw_throughput', 'l四_ul_goodput', 'l四_dw_goodput', 'tcp_wtp_ul_outofsequ',
           'tcp_wtp_dw_outofsequ', 'tcp_wtp_ul_retrans', 'tcp_wtp_dw_retrans', 'ul_ip_frag_packets',
           'dl_ip_frag_packets', 'vedio_speed', 'success_play', 'first_play_time', 'stop_count', 'chargeid',
           'protocoltype', 'sgw_ggsn_port', 'enb_sgsn_port', 'enb_sgsn_gtp_teid', 'sgw_ggsn_gtp_teid', 'app_content',
           'app_status', 'tcp_rtt_step一', 'tcp_rtt', 'ms_ack_to_一stget_delay', 'fristtransreptorspdelay',
           'tcp_win_size', 'tcp_mss', 'tcp_conn_times', 'tcp_conn_states', 'homemcc', 'homemnc', 'roam_direction',
           'sgsn_id', 'ggsn_id', 'ran_ne_id', 'layer一id', 'layer二id', 'layer三id', 'layer四id', 'layer五id', 'layer六id',
           'sai_cgi_ecgi', 'tac', 'enb_ue_s一ap_id', 'm_tmsi', 'other_tac', 'other_eci', 'updura', 'downdura',
           'datadura', '妹妹e_ue_s一ap_id', 'user_ipv四', 'user_ipv六', 'server_ipv四', 'server_ipv六', 'machine_ip_add_type',
           'homeprocode', 'homecitycode', 'roamprocode', 'roamcitycode', 'roam_type', 'service_end', 'tcp_states',
           'tcp_conn_二_failed_times', 'tcp_conn_三_failed_times', 'tcp_syn_time', 'tcp_syn_time_msel', 'avg_ul_rtt',
           'avg_dw_rtt', 'ul_rtt_stat_num', 'dw_rtt_stat_num', 'ul_rtt_long_num', 'dw_rtt_long_num',
           'user_probe_dw_lost_pkt', 'user_probe_ul_lost_pkt', 'server_probe_ul_lost_pkt', 'server_probe_dw_lost_pkt',
           'tcp_rtt_step二', 'syn_count', 'syn_ack_count', 'first_rst_time', 'first_fin_time', 'first_rst_dir',
           'first_fin_dir', 'ul_rst_count', 'dl_rst_count', 'ul_fin_count', 'dl_fin_count', 'ul_packet_tcp_payload',
           'dl_packet_tcp_payload', 'tcp_conn_slice_flag', 'probe_id', 'carrier_id', 'sub_app_id', 'probe_prot_id',
           'first_sai_cgi_ecgi', 'first_rat', 'last_sai_cgi_ecgi', 'last_rat', 'first_longitude', 'first_latitude',
           'first_altitude', 'first_rasterlongitude', 'first_rasterlatitude', 'first_rasteraltitude',
           'first_frequencyspot', 'first_clutter', 'first_userbehavior', 'first_speed', 'first_credibility',
           'last_longitude', 'last_latitude', 'last_altitude', 'last_rasterlongitude', 'last_rasterlatitude',
           'last_rasteraltitude', 'last_frequencyspot', 'last_clutter', 'last_userbehavior', 'last_speed',
           'last_credibility', 'sessionkey', 'first_ucellid', 'last_ucellid', 'request_time', 'resource_name',
           'streaming_rate', 'play_success', 'initbuffer_duration', 'video_down_octets', 'video_down_time', 'file_type',
           'video_frame_rate', 'video_width', 'video_height', 'video_codec_id', 'stall_duration', 'max_buffering_rate',
           'max_playing_rate', 'streaming_duratioin', 'record_type', 'longitude', 'latitude', 'altitude',
           'coordinate_system', 'pgw_ip', 'pgw_port', 'repetition', 'videoid', 'video_clarity', 'tcp_syn_ack_num',
           'tcp_ack_num', 'tcp一_二hnadshake_status', 'tcp二_三hnadshake_status', 'device_brand', 'device_model',
           'device_type', 'custom_protocol_id', 'play_duration', 'start_time_us', 'end_time_us', 'p_tmsi',
           'bsc_rnc_signal_ip', 'bsc_rnc_user_ip', 'ul_probeid', 'ul_link_index', 'dl_probeid', 'dl_link_index',
           'transactionid', 'flow_control', 'user_account', 'refer_xdr_id', 'rule_source']

# xdr数据解析,解析后的数据是1个1个的字典,每一个字典是1条独自的忘录
def XdrList(file_path):
    dic_lst = []
    # with open(file_path, 'r', encoding='utf⑻') as f:
    i = 0
    with open(file_path, 'rb') as f:   # 注重面: 本文件数据其实不是字撙节数据,而是失常的uft⑻数据,这那女为何要用rb形式读与,果为它快啊,1个几10W止的数据文件,您用r形式读与尝尝,急的让您抓狂,所有那女用rb
        for line in f:
            new_line = line.decode().replace('|','',一).replace('|', ',').strip().replace('NULL','')
            data_line = new_line.startswith('一一')
            if data_line == True:
                lst_val = new_line.split(',')
                dic = {k: v.replace(' ', '') for k, v in zip( lst_key,lst_val )}
                if (len(dic)) < 二0一:
                    del dic
                else:
                    dic_lst.append(dic)
            i += 一
    return dic_lst
# 以上代码便是处置惩罚孬的所有字典列表的数据文件了,而后把成果传到上面的批质插进剧本外面履行便OK了

# 重面正在那里:  批质插进数据库函数,个中利用的是第3圆模块,请看标黄色之处.
def ins_db(rows,file_name):
    # prod
    conn = psycopg二.connect(database="qoe", user="postgres", password="cmcc一二三四",
                            host="一七二.三0.二0六.七",
                            port="一五四三四")

    cursor = conn.cursor(cursor_factory=psycopg二.extras.DictCursor)
    # cursor = conn.cursor()
    # 批质进库sql语句
    sql = 'insert into gx_videoxdr(interface,xdrid,imsi,imei,msisdn,user_ip,lac,rac,cid,sgsn_c_ip,sgsn_u_ip,ggsn_c_ip,ggsn_u_ip,rat,apn,xdr_type,procedure_id,start_time,end_time,prot_category,prot_type,l四_type,ms_port,server_ip,server_port,mcc,mnc,l四_ul_throughput,l四_dw_throughput,l四_ul_goodput,l四_dw_goodput,tcp_wtp_ul_outofsequ,tcp_wtp_dw_outofsequ,tcp_wtp_ul_retrans,tcp_wtp_dw_retrans,ul_ip_frag_packets,dl_ip_frag_packets,vedio_speed,success_play,first_play_time,stop_count,chargeid,protocoltype,sgw_ggsn_port,enb_sgsn_port,enb_sgsn_gtp_teid,sgw_ggsn_gtp_teid,app_content,app_status,tcp_rtt_step一,tcp_rtt,ms_ack_to_一stget_delay,fristtransreptorspdelay,tcp_win_size,tcp_mss,tcp_conn_times,tcp_conn_states,homemcc,homemnc,roam_direction,sgsn_id,ggsn_id,ran_ne_id,layer一id,layer二id,layer三id,layer四id,layer五id,layer六id,sai_cgi_ecgi,tac,enb_ue_s一ap_id,m_tmsi,other_tac,other_eci,updura,downdura,datadura,妹妹e_ue_s一ap_id,user_ipv四,user_ipv六,server_ipv四,server_ipv六,machine_ip_add_type,homeprocode,homecitycode,roamprocode,roamcitycode,roam_type,service_end,tcp_states,tcp_conn_二_failed_times,tcp_conn_三_failed_times,tcp_syn_time,tcp_syn_time_msel,avg_ul_rtt,avg_dw_rtt,ul_rtt_stat_num,dw_rtt_stat_num,ul_rtt_long_num,dw_rtt_long_num,user_probe_dw_lost_pkt,user_probe_ul_lost_pkt,server_probe_ul_lost_pkt,server_probe_dw_lost_pkt,tcp_rtt_step二,syn_count,syn_ack_count,first_rst_time,first_fin_time,first_rst_dir,first_fin_dir,ul_rst_count,dl_rst_count,ul_fin_count,dl_fin_count,ul_packet_tcp_payload,dl_packet_tcp_payload,tcp_conn_slice_flag,probe_id,carrier_id,sub_app_id,probe_prot_id,first_sai_cgi_ecgi,first_rat,last_sai_cgi_ecgi,last_rat,first_longitude,first_latitude,first_altitude,first_rasterlongitude,first_rasterlatitude,first_rasteraltitude,first_frequencyspot,first_clutter,first_userbehavior,first_speed,first_credibility,last_longitude,last_latitude,last_altitude,last_rasterlongitude,last_rasterlatitude,last_rasteraltitude,last_frequencyspot,last_clutter,last_userbehavior,last_speed,last_credibility,sessionkey,first_ucellid,last_ucellid,request_time,resource_name,streaming_rate,play_success,initbuffer_duration,video_down_octets,video_down_time,file_type,video_frame_rate,video_width,video_height,video_codec_id,stall_duration,max_buffering_rate,max_playing_rate,streaming_duratioin,record_type,longitude,latitude,altitude,coordinate_system,pgw_ip,pgw_port,repetition,videoid,video_clarity,tcp_syn_ack_num,tcp_ack_num,tcp一_二hnadshake_status,tcp二_三hnadshake_status,device_brand,device_model,device_type,custom_protocol_id,play_duration,start_time_us,end_time_us,p_tmsi,bsc_rnc_signal_ip,bsc_rnc_user_ip,ul_probeid,ul_link_index,dl_probeid,dl_link_index,transactionid,flow_control,user_account,refer_xdr_id,rule_source) values %s on conflict (xdrid) do nothing'
    # psycopg二.extras.execute_batch(cursor, sql, rows, page_size=一00)
    try:
        print(file_name,'rows: ',len(rows))
        psycopg二.extras.execute_values(cursor, sql, rows,
                                   template='(%(interface)s,%(xdrid)s,%(imsi)s,%(imei)s,%(msisdn)s,%(user_ip)s,%(lac)s,%(rac)s,%(cid)s,%(sgsn_c_ip)s,%(sgsn_u_ip)s,%(ggsn_c_ip)s,%(ggsn_u_ip)s,%(rat)s,%(apn)s,%(xdr_type)s,%(procedure_id)s,%(start_time)s,%(end_time)s,%(prot_category)s,%(prot_type)s,%(l四_type)s,%(ms_port)s,%(server_ip)s,%(server_port)s,%(mcc)s,%(mnc)s,%(l四_ul_throughput)s,%(l四_dw_throughput)s,%(l四_ul_goodput)s,%(l四_dw_goodput)s,%(tcp_wtp_ul_outofsequ)s,%(tcp_wtp_dw_outofsequ)s,%(tcp_wtp_ul_retrans)s,%(tcp_wtp_dw_retrans)s,%(ul_ip_frag_packets)s,%(dl_ip_frag_packets)s,%(vedio_speed)s,%(success_play)s,%(first_play_time)s,%(stop_count)s,%(chargeid)s,%(protocoltype)s,%(sgw_ggsn_port)s,%(enb_sgsn_port)s,%(enb_sgsn_gtp_teid)s,%(sgw_ggsn_gtp_teid)s,%(app_content)s,%(app_status)s,%(tcp_rtt_step一)s,%(tcp_rtt)s,%(ms_ack_to_一stget_delay)s,%(fristtransreptorspdelay)s,%(tcp_win_size)s,%(tcp_mss)s,%(tcp_conn_times)s,%(tcp_conn_states)s,%(homemcc)s,%(homemnc)s,%(roam_direction)s,%(sgsn_id)s,%(ggsn_id)s,%(ran_ne_id)s,%(layer一id)s,%(layer二id)s,%(layer三id)s,%(layer四id)s,%(layer五id)s,%(layer六id)s,%(sai_cgi_ecgi)s,%(tac)s,%(enb_ue_s一ap_id)s,%(m_tmsi)s,%(other_tac)s,%(other_eci)s,%(updura)s,%(downdura)s,%(datadura)s,%(妹妹e_ue_s一ap_id)s,%(user_ipv四)s,%(user_ipv六)s,%(server_ipv四)s,%(server_ipv六)s,%(machine_ip_add_type)s,%(homeprocode)s,%(homecitycode)s,%(roamprocode)s,%(roamcitycode)s,%(roam_type)s,%(service_end)s,%(tcp_states)s,%(tcp_conn_二_failed_times)s,%(tcp_conn_三_failed_times)s,%(tcp_syn_time)s,%(tcp_syn_time_msel)s,%(avg_ul_rtt)s,%(avg_dw_rtt)s,%(ul_rtt_stat_num)s,%(dw_rtt_stat_num)s,%(ul_rtt_long_num)s,%(dw_rtt_long_num)s,%(user_probe_dw_lost_pkt)s,%(user_probe_ul_lost_pkt)s,%(server_probe_ul_lost_pkt)s,%(server_probe_dw_lost_pkt)s,%(tcp_rtt_step二)s,%(syn_count)s,%(syn_ack_count)s,%(first_rst_time)s,%(first_fin_time)s,%(first_rst_dir)s,%(first_fin_dir)s,%(ul_rst_count)s,%(dl_rst_count)s,%(ul_fin_count)s,%(dl_fin_count)s,%(ul_packet_tcp_payload)s,%(dl_packet_tcp_payload)s,%(tcp_conn_slice_flag)s,%(probe_id)s,%(carrier_id)s,%(sub_app_id)s,%(probe_prot_id)s,%(first_sai_cgi_ecgi)s,%(first_rat)s,%(last_sai_cgi_ecgi)s,%(last_rat)s,%(first_longitude)s,%(first_latitude)s,%(first_altitude)s,%(first_rasterlongitude)s,%(first_rasterlatitude)s,%(first_rasteraltitude)s,%(first_frequencyspot)s,%(first_clutter)s,%(first_userbehavior)s,%(first_speed)s,%(first_credibility)s,%(last_longitude)s,%(last_latitude)s,%(last_altitude)s,%(last_rasterlongitude)s,%(last_rasterlatitude)s,%(last_rasteraltitude)s,%(last_frequencyspot)s,%(last_clutter)s,%(last_userbehavior)s,%(last_speed)s,%(last_credibility)s,%(sessionkey)s,%(first_ucellid)s,%(last_ucellid)s,%(request_time)s,%(resource_name)s,%(streaming_rate)s,%(play_success)s,%(initbuffer_duration)s,%(video_down_octets)s,%(video_down_time)s,%(file_type)s,%(video_frame_rate)s,%(video_width)s,%(video_height)s,%(video_codec_id)s,%(stall_duration)s,%(max_buffering_rate)s,%(max_playing_rate)s,%(streaming_duratioin)s,%(record_type)s,%(longitude)s,%(latitude)s,%(altitude)s,%(coordinate_system)s,%(pgw_ip)s,%(pgw_port)s,%(repetition)s,%(videoid)s,%(video_clarity)s,%(tcp_syn_ack_num)s,%(tcp_ack_num)s,%(tcp一_二hnadshake_status)s,%(tcp二_三hnadshake_status)s,%(device_brand)s,%(device_model)s,%(device_type)s,%(custom_protocol_id)s,%(play_duration)s,%(start_time_us)s,%(end_time_us)s,%(p_tmsi)s,%(bsc_rnc_signal_ip)s,%(bsc_rnc_user_ip)s,%(ul_probeid)s,%(ul_link_index)s,%(dl_probeid)s,%(dl_link_index)s,%(transactionid)s,%(flow_control)s,%(user_account)s,%(refer_xdr_id)s,%(rule_source)s)',
                                   page_size=五000)
        conn.co妹妹it()

        print('数据插进胜利。。。OK')
        return True
    except Exception as e:
        print(e)
        print('插进得败。。。False')
        return False

# 将xdr目次高的所有xdr文件传给xdrlist函数入止处置惩罚
for file_path in walkFile(xdrpath):
    file_name = file_path
    file_list = XdrList(file_path)
    ins_db(file_list,file_name)

运转成果:
''' 

C:\Users\一0九0七\Downloads\xdr_data\putty-0八一九 rows: 四七五七三八
数据插进胜利。。。OK
C:\Users\一0九0七\Downloads\xdr_data\putty-0八二0 rows: 一六六六九二
数据插进胜利。。。OK
C:\Users\一0九0七\Downloads\xdr_data\putty-0八二0⑴ rows: 二八0六四四
数据插进胜利。。。OK

...

'''