利用redis协助mysql数据库搬迁

最近公司新项目上线，需要数据库搬迁，但新版本和老版本数据库差距比较大，关系也比较复杂。如果用传统办法，需要撰写很多mysql脚本，工程量虽然不大，但对于没有dba的公司来说，稍微有点难度。本人就勉为其难，用redis作为mysql中转站，先把原来的mysql数据转移到redis里面，再从redis转换到新的mysql。整个过程脉络清晰，脚本简单。

首先，公司之前的项目是，flask+sqlalchemy的方式，直接复制原来的model文件，稍微修改一下，就有以下代码。

   1 # coding:utf-8   2 from sqlalchemy import create_engine, ForeignKey, Column, Integer, String, Text, DateTime, Boolean, and_, or_,/   3     SmallInteger, func, Numeric, select, Float, Table, TIMESTAMP, DECIMAL, desc   4 from sqlalchemy.orm import relationship, backref, sessionmaker, scoped_session   5 from sqlalchemy.ext.declarative import declarative_base   6 from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method, Comparator   7 from sqlalchemy import event   8 import redis   9 import json  10   11 mysql_info = "mysql://username:password@ip_address:3306/databasename?charset=utf8"  12 redis_store = redis.Redis(host='127.0.0.1', port=6380, password='password', db=5)  13   14 engine = create_engine(mysql_info, pool_recycle=7200)  15 Base = declarative_base()  16   17   18 class User(Base):  19     __tablename__ = 'user'  20   21     id = Column('id', Integer, primary_key=True)  22     phone_number = Column('phone_number', String(11), index=True)  23     password = Column('password', String(30))  24     nickname = Column('nickname', String(30), index=True, nullable=True)  25     headpic = Column('headpic', String(100), nullable=True)  26     register_time = Column('register_time', DateTime, index=True)  27     email = Column('email', String(50), default='', index=True)  28     real_name = Column('real_name', String(50), default='', index=True)  29     user_admin = Column('user_admin', Integer, index=True, default=0)  30     rc_token = Column('rc_token', String(100))  31     device_token = Column('device_token', String(100), unique=True, nullable=True)  32     unit_id = Column('unit_id', Integer, ForeignKey('unit.id'))  33     unit = relationship("Unit", backref=backref('users'))  34   35     def to_dict(self):  36         return dict(id=self.id, phone_number=self.phone_number,  37                     password=self.password, nickname=self.nickname,  38                     headpic=self.headpic, register_time=self.register_time.strftime('%Y-%m-%d %H:%M:%S'),  39                     user_admin=self.user_admin, rc_token=self.rc_token,  40                     unit_id=self.unit_id, device_token=self.device_token,  41                     )  42   43   44 class Group(Base):  45     __tablename__ = 'groups'  46   47     id = Column('id', Integer, primary_key=True)  48     name = Column('name', String(100), index=True)  49     address = Column('address', String(100), index=True, nullable=True)  50     contact_person = Column('contactperson', String(30), nullable=True)  51     contact_number = Column('contactnumber', String(30), nullable=True)  52     unit_id = Column('unit_id', Integer, ForeignKey('unit.id'))  53     unit = relationship('Unit', backref=backref('groups'))  54   55   56 class Monitor(Base):  57     __tablename__ = 'monitor'  58   59     id = Column('id', Integer, primary_key=True)  60     u_id = Column('sn_num', String(10), index=True)  61     phone_num = Column('phone_num', String(20), index=True, nullable=True)  62     name = Column('name', String(40), index=True, nullable=True)  63     position = Column('position', String(40), nullable=True)  64     join_time = Column('join_time', DateTime, index=True, nullable=True)  65     group_id = Column('group_id', Integer, ForeignKey('groups.id'))  66     group = relationship("Group", backref=backref('monitors'))  67     longitude = Column('longitude', DECIMAL(12, 9), default=31.000000)  68     latitude = Column('latitude', DECIMAL(12, 9), default=121.000000)  69   70     def to_dict(self):  71         unit_id = self.group.unit_id  72         return dict(id=self.id, u_id=self.u_id, phone_num=self.phone_num, name=self.name,  73                     position=self.position, join_time=self.join_time.strftime('%Y-%m-%d %H:%M:%S'), unit_id=unit_id,  74                     longitude=str(self.longitude), latitude=str(self.latitude))  75   76   77 class Unit(Base):  78     __tablename__ = 'unit'  79   80     id = Column('id', Integer, primary_key=True)  81     name = Column('name', String(100), index=True, nullable=True)  82     address = Column('address', String(100), index=True, nullable=True)  83     contact_person = Column('contactperson', String(30), index=True, nullable=True)  84     contact_number = Column('contactnumber', String(30), nullable=True)  85     device_operation_password = Column('device_operation_password', String(4), nullable=True)  86     rc_group_id = Column('rc_group_id', String(50), index=True, nullable=True)  87     rc_group_name = Column('rc_group_name', String(50), index=True, nullable=True)  88     longitude = Column('longitude', DECIMAL(12, 9), nullable=True)  89     latitude = Column('latitude', DECIMAL(12, 9), nullable=True)  90     active = Column('active', SmallInteger, index=True, default=0)  91   92     def to_dict(self):  93         return dict(id=self.id, name=self.name, address=self.address,  94                     contact_person=self.contact_person, contact_number=self.contact_number,  95                     device_operation_password=self.device_operation_password,  96                     rc_group_id=self.rc_group_id, rc_group_name=self.rc_group_name,  97                     longitude=str(self.longitude), latitude=str(self.latitude), active=self.active)  98   99 db_session = scoped_session(sessionmaker(autocommit=False, 100                                          autoflush=False, 101                                          bind=engine)) 102  103 Base.query = db_session.query_property() 104  105  106 def old_sql_2_redis(): 107     redis_store.flushdb() 108     units = Unit.query.all() 109     [redis_store.rpush('units', json.dumps(unit.to_dict())) for unit in units] 110  111     users = User.query.all() 112     [redis_store.rpush('users', json.dumps(user.to_dict())) for user in users] 113  114     monitors = Monitor.query.all() 115     [redis_store.rpush('monitors', json.dumps(monitor.to_dict())) for monitor in monitors] 116  117 if __name__ == '__main__': 118     old_sql_2_redis() 

把原来的数据库复制到本地redis了，看查看里面的元素。比如user数据，直接建一个user的list，每个appened其json格式的数据就可以了。主要注意点就是，如果新版本的表中没有的字段，就不要写入的redis了。

第二步就是把本地redis里面的数据，放到新的数据库上。这边要注意以下，把需要修改的地方要标示出来，可能加了其他字段，这些字段没有默认值的话，要自己添加上。过程也比较简单。

   1 # coding:utf-8   2 from sqlalchemy import create_engine, ForeignKey, Column, Integer, String, Text, DateTime, Boolean, and_, or_,/   3     SmallInteger, func, Numeric, select, Float, Table, TIMESTAMP, DECIMAL, desc   4 from sqlalchemy.orm import relationship, backref, sessionmaker, scoped_session   5 from sqlalchemy.ext.declarative import declarative_base   6 from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method, Comparator   7 from sqlalchemy import event   8 import redis   9 import json  10 import datetime  11 import uuid  12   13 mysql_info = "mysql://username:password@ip_address:3306/database_name?charset=utf8"  14 redis_store = redis.Redis(host='127.0.0.1', port=6380, password='password', db=5)  15   16 engine = create_engine(mysql_info, pool_recycle=7200)  17 Base = declarative_base()  18   19   20 class User(Base):  21     __tablename__ = 'user'  22   23     id = Column('id', Integer, primary_key=True)  24     phone_number = Column('phone_number', String(11), index=True)  25     password = Column('password', String(30))  26     nickname = Column('nickname', String(30), index=True, nullable=True)  27     headpic = Column('headpic', String(100), nullable=True)  28     register_time = Column('register_time', DateTime, index=True, default=datetime.datetime.now)  29     user_admin = Column('user_admin', SmallInteger, index=True, default=0)  30     hidden_user = Column('hidden_user', SmallInteger, index=True, default=0)  31     rc_token = Column('rc_token', String(100), nullable=True)  32     device_token = Column('device_token', String(100), unique=True, nullable=True)  33     unit_id = Column('unit_id', Integer, ForeignKey('unit.id'))  34     unit = relationship('Unit', backref=backref('users'))  35   36   37 class Monitor(Base):  38     __tablename__ = 'monitor'  39   40     id = Column('id', Integer, primary_key=True)  41     # device_type 1代表灭弧 2代表电气火灾  42     device_type = Column('device_type', SmallInteger, index=True, default=1)  43     u_id = Column('sn_num', String(10), index=True)  44     phone_num = Column('phone_num', String(20), index=True, nullable=True)  45     name = Column('name', String(40), index=True)  46     position = Column('position', String(40), nullable=True)  47     join_time = Column('join_time', DateTime, index=True)  48     longitude = Column('longitude', DECIMAL(12, 9), default=31.000000)  49     latitude = Column('latitude', DECIMAL(12, 9), default=121.000000)  50     unit_id = Column('unit_id', Integer, ForeignKey('unit.id'))  51   52   53 class Unit(Base):  54     __tablename__ = 'unit'  55   56     id = Column('id', Integer, primary_key=True)  57     name = Column('name', String(100), index=True)  58     address = Column('address', String(100), index=True, nullable=True)  59     contact_person = Column('contactperson', String(30), nullable=True)  60     contact_number = Column('contactnumber', String(30), nullable=True)  61     device_operation_password = Column('device_operation_password', String(4), default='1234')  62     rc_group_id = Column('rc_group_id', String(36), default=str(uuid.uuid1()))  63     rc_group_name = Column('rc_group_name', String(50), nullable=True)  64     longitude = Column('longitude', DECIMAL(12, 9), default=31.000000)  65     latitude = Column('latitude', DECIMAL(12, 9), default=121.000000)  66     active = Column('active', SmallInteger, index=True, default=0)  67     is_group = Column('is_group', SmallInteger, index=True, default=0)  68     parent_id = Column('parent_id', Integer, ForeignKey('unit.id'))  69   70   71 db_session = scoped_session(sessionmaker(autocommit=False,  72                                          autoflush=False,  73                                          bind=engine))  74   75 Base.query = db_session.query_property()  76   77   78 def redis_2_new_sql():  79     units_json = redis_store.lrange('units', 0, -1)  80     units = [Unit(is_group=0, parent_id=None, **json.loads(unit_json)) for unit_json in units_json]  81     [db_session.add(unit) for unit in units]  82     try:  83         db_session.commit()  84     except Exception as e:  85         print e  86         db_session.rollback()  87         return  88   89     users_json = redis_store.lrange('users', 0, -1)  90     users = [User(hidden_user=0, **json.loads(uer_json)) for uer_json in users_json]  91     [db_session.add(user) for user in users]  92     try:  93         db_session.commit()  94     except Exception as e:  95         print e  96         db_session.rollback()  97         return  98   99     monitors_json = redis_store.lrange('monitors', 0, -1) 100     monitors = [Monitor(**json.loads(monitor_json)) for monitor_json in monitors_json] 101     [db_session.add(monitor) for monitor in monitors] 102     try: 103         db_session.commit() 104     except Exception as e: 105         print e 106         db_session.rollback() 107         return 108  109 if __name__ == '__main__': 110     redis_2_new_sql() 

整个过程相当简单，需要注意的就是，前后顺序，因为user和monitor都有一个外键指向unit，所以要先恢复unit表，这样就不会出错了。

看来以后我有新方法搬迁数据库了。