1.8亿条海量Txt数据存储MySQL实践

公众号guangcity

发布于 2019-09-20 17:25:14

2.2K00

代码可运行

文章被收录于专栏：光城(guangcity)光城(guangcity)

运行总次数：0

代码可运行

0.导语

最近出去旅游了，嗨皮了嗨皮，明天上班，开始做作业，今日将1.8亿数据存储的方式进行总结，欢迎大家拍砖！

预告：后面推送大数据伪分布式从零搭建到1.8亿海量数据从Mysql至HBase数据转存技术分析与应用！

1.搭建MySQL数据库

电脑环境为Ubuntu16.04系统。

1.1 MySQL安装

sudo apt-get install mysql-server

出现依赖问题：

sudo apt-get install -f

最后再次运行上述命令即可。

1.2 配置目录

安装成功后，相关配置文件如下：

数据库目录：/var/lib/mysql/
配置文件：/usr/share/mysql（命令及配置文件），/etc/mysql（如：my.cnf）
相关命令：/usr/bin(mysqladmin mysqldump等命令) 和/usr/sbin
启动脚本：/etc/init.d/mysql（启动脚本文件mysql的目录）

可以通过进入上述相关目录查看，例如：查看数据库：

查看配置文件：

1.3 服务管理

启动MySQL：sudo service mysql start

进入MySQL：mysql -uroot -p

#启动
sudo service mysql start
#停止
sudo service mysql stop
#服务状态
sudo service mysql status

2.导入海量GPS数据

选择导入方式为：load data命令行方式与程序插入方式。

导入数据之前，根据字段描述编写SQL语句进行创建数据库与表操作。

字段描述：

数据以ASCII文本表示，以逗号为分隔符，以回车换行符(0x0D 0x0A)结尾。数据项及顺序：车辆标识、触发事件、运营状态、GPS时间、GPS经度、GPS纬度,、GPS速度、GPS方向、GPS状态

车辆标识：6个字符

触发事件：0=变空车，1=变载客，2=设防，3=撤防，4=其它

运营状态：0=空车，1=载客，2=驻车，3=停运，4=其它

GPS时间：格式yyyymmddhhnnss，北京时间

GPS经度：格式ddd.ddddddd，以度为单位。

GPS纬度：格式dd.ddddddd，以度为单位。

GPS速度：格式ddd，取值000-255内整数，以公里/小时为单位。

GPS方位：格式ddd，取值000-360内整数，以度为单位。

GPS状态：0=无效，1=有效

结束串:回车符+换行符

创建数据库

create database loaddb;

创建数据库表

创建load data方式表

CREATE TABLE loadTable(id int primary key not null auto_increment,
carflag VARCHAR(6),touchevent CHAR(1),opstatus CHAR(1),gpstime DATETIME,
gpslongitude DECIMAL(10,7),gpslatitude DECIMAL(9,7),gpsspeed TINYINT,
gpsorientation SMALLINT,gpsstatus CHAR(1));

创建程序插入法表

CREATE TABLE loadTable1(id int primary key not null auto_increment,
carflag VARCHAR(6),touchevent CHAR(1),opstatus CHAR(1),gpstime DATETIME,
gpslongitude DECIMAL(10,7),gpslatitude DECIMAL(9,7),gpsspeed TINYINT,
gpsorientation SMALLINT,gpsstatus CHAR(1));

2.1 load data命令行方式

导入数据

load data local infile "/home/light/mysql/gps1.txt" into table loadTable fields terminated by ',' lines terminated by "\n"  (carflag, touchevent, opstatus,gpstime,gpslongitude,gpslatitude,gpsspeed,gpsorientation,gpsstatus);

完成耗时：

2.2 程序插入法

这里使用Python操纵MySQL数据库，进行SQL的插入。

基本思路是使用Python的pymysql对MySQL进行连接，使用executemany进行批量提交，每隔7万提交1次。

Python代码：

import pymysql
import time

class MyPyMysql:
    def __init__(self, host, port, username, password, db, charset='utf8'):
        self.host = host  # mysql主机地址
        self.port = port  # mysql端口
        self.username = username  # mysql远程连接用户名
        self.password = password  # mysql远程连接密码
        self.db = db  # mysql使用的数据库名
        self.charset = charset  # mysql使用的字符编码,默认为utf8
        self.pymysql_connect()  # __init__初始化之后，执行的函数

    def pymysql_connect(self):
        # pymysql连接mysql数据库
        # 需要的参数host,port,user,password,db,charset
        self.conn = pymysql.connect(host=self.host,
                                    port=self.port,
                                    user=self.username,
                                    password=self.password,
                                    db=self.db,
                                    charset=self.charset
                                    )
        # 连接mysql后执行的函数
        self.run()

    def run(self):
        # 创建游标
        self.cur = self.conn.cursor()

        # 定义sql语句,插入数据id,name,gender,email
        sql = "insert into loadTable1(carflag, touchevent, opstatus,gpstime,gpslongitude,gpslatitude,gpsspeed,gpsorientation,gpsstatus) values(%s,%s,%s,str_to_date(%s,'%%Y-%%m-%%d %%H:%%i:%%s'),%s,%s,%s,%s,%s)"

        # 定义总插入行数为一个空列表
        data_list = []
        count = 0
        with open('/home/light/mysql/gps1.txt', 'r') as fp:
            for line in fp:
                line = line.split(',')
                # print(line)
                carflag, touchevent, opstatus, gpstime, gpslongitude, gpslatitude, gpsspeed, gpsorientation, gpsstatu = line[0],line[1],line[2],line[3],line[4],line[5],line[6],line[7],line[8].strip()

                gpstime = gpstime[:4]+'-'+gpstime[4:6]+'-'+gpstime[6:8]+' '+gpstime[8:10]+':'+gpstime[10:12]+':'+gpstime[12:14]
                tup = (carflag, touchevent, opstatus, gpstime, gpslongitude, gpslatitude, gpsspeed, gpsorientation, gpsstatu)
                data_list.append(tup)
                count += 1
                if count and count%70000==0:
                    # 执行多行插入，executemany(sql语句,数据(需一个元组类型))
                    self.cur.executemany(sql, data_list)
                    # 提交数据,必须提交，不然数据不会保存
                    self.conn.commit()
                    data_list = []
                    print("提交了：" + str(count) + "条数据")

        if data_list:
            # 执行多行插入，executemany(sql语句,数据(需一个元组类型))
            self.cur.executemany(sql, data_list)
            # 提交数据,必须提交，不然数据不会保存
            self.conn.commit()
            print("提交了：" + str(count) + "条数据")
        self.cur.close()  # 关闭游标
        self.conn.close()  # 关闭pymysql连接

if __name__ == '__main__':
    start_time = time.time()  # 计算程序开始时间
    st = MyPyMysql('127.0.0.1', 3306, 'root', 'xxxx', 'loaddb')  # 实例化类，传入必要参数
    print('程序耗时{:.2f}'.format(time.time() - start_time))  # 计算程序总耗时

完成耗时：

3.分析两种插入方式

两者异同比较

相同点	不同点
两者都是通过读取本地txt文件，按照相同的分隔来读取进行插入。	程序插入法实质为insert语句间接执行。load data设计用于在单个操作中大量加载表格数据。

两者效率比较

两者耗时如下：

第一种：load data

用时1h11分。

第二种：程序插入法

用时：27322.45/36=7.58h

上述对比可知，load data效率非常高，原因在于使用的是load data infile方式，而第二种则为传统的insert方式。

究其根源主要是MySQL内部对于load 和 insert的处理机制不同。

Load的处理机制是:在执行load之前,会关掉索引,当load全部执行完成后,再重新创建索引.

Insert的处理机制是:每插入一条则更新一次数据库,更新一次索引.

另外,load与insert的不同还体现在load省去了sql语句解析,sql引擎处理,而是直接生成文件数据块,所以会比Insert快很多.

4.出租车轨迹数据的分析

4.1 统计总记录数、统计出租车数量

统计：

select count(*) AS '总记录数', count(distinct carflag) AS '出租车数量' from loadTable;

4.2 编写3个SQL语句

查询前200万数据中各个触发事件的总数并降序排序。

select touchevent,count(id) as number from (select * from loadTable as lt  limit 2000000) as t group by touchevent order by number desc;

查询前200万数据中GPS速度不小于30时公里/小时中各个运营状态的总数并按照升序排序。

select t.opstatus,count(t.id) as number from (select * from loadTable as lt  limit 2000000) as t where t.gpsspeed>=30  group by t.opstatus order by number;

查询前200万数据GPS时间在2012-11-01 00:35中触发事件为变其他且运营状态为载客状态下的总信息数。

select count(*) from (select * from loadTable as lt  limit 2000000) as t where t.touchevent=4 and t.opstatus=1 and t.gpstime like '2012-11-01 00:35:%';

4.3 数据库连接

这里使用Python完成本题。

连接数据库

class analysisData:
    def __init__(self, host, port, username, password, db, charset='utf8'):
        self.host = host  # mysql主机地址
        self.port = port  # mysql端口
        self.username = username  # mysql远程连接用户名
        self.password = password  # mysql远程连接密码
        self.db = db  # mysql使用的数据库名
        self.charset = charset  # mysql使用的字符编码,默认为utf8
        self.pymysql_connect()  # __init__初始化之后，执行的函数

    def pymysql_connect(self):
        # pymysql连接mysql数据库
        # 需要的参数host,port,user,password,db,charset
        self.conn = pymysql.connect(host=self.host,
                                    port=self.port,
                                    user=self.username,
                                    password=self.password,
                                    db=self.db,
                                    charset=self.charset
                                    )
        # 连接mysql后执行的函数
        self.run()

获取数据

def run(self):
    # 创建游标
    self.cur = self.conn.cursor()

    touchsql = "select touchevent,count(id) as number from loadTable group by touchevent;"
    self.cur.execute(touchsql)

    # 查询数据库多条数据
    result = self.cur.fetchall()
    event = list(zip(*list(result)))[1]
    print(result)

    opssql = "select opstatus,count(id) as number from loadTable group by opstatus;"

    self.cur.execute(opssql)

    # 查询数据库多条数据
    result = self.cur.fetchall()
    status = list(zip(*list(result)))[1]
    print(result)
    for data in result:
        print(data)

    self.visualCategory(event,status)

    self.cur.close()  # 关闭游标
    self.conn.close()  # 关闭pymysql连接

数据分析可视化

def visualCategory(self,event,status):
    bar = (
        Bar(init_opts=opts.InitOpts(theme=ThemeType.LIGHT))
            .add_xaxis([0, 1, 2, 3, 4])
            .add_yaxis("触发事件", event)
            .add_yaxis("运营状态", status)
            .set_global_opts(title_opts=opts.TitleOpts(title="触发事件与运营状态分类统计图"))
    )
    bar.render()

类实例化与调用

if __name__ == '__main__':
    start_time = time.time()  # 计算程序开始时间
    st = analysisData('127.0.0.1', 3306, 'root', 'xxxx', 'loaddb')  # 实例化类，传入必要参数
    print('程序耗时{:.2f}'.format(time.time() - start_time))  # 计算程序总耗时

数据分析图：