sqlalchemy更新数据与原生sql使用case when 更新

我就是我 2022-11-25 10:04 347阅读 0赞

最近有个任务，将表中的存储的首拼改为全拼，比如infos表中有name字段为中文，在name\_acronym字段中存储着name的首拼即拼音的首字母，现在需要将其改为全拼。  
表中大概有一百多万数据，一开始用的sqlalchemy，但是速度太慢，并且没跑完可能脚本就被kill了，所以后来改为了原生sql，下面记录一下我的**完成过程以及踩过的坑**。

生成全拼使用的是slug方法，在pypinyin库中的core.py中定义了slug函数，下面是官方的定义：

def slug(hans, style=Style.NORMAL, heteronym=False, separator='-',
             errors='default', strict=True):
        """生成 slug 字符串. :param hans: 汉字 :type hans: unicode or list :param style: 指定拼音风格，默认是 :py:attr:`~pypinyin.Style.NORMAL` 风格。 更多拼音风格详见 :class:`~pypinyin.Style` :param heteronym: 是否启用多音字 :param separstor: 两个拼音间的分隔符/连接符 :param errors: 指定如何处理没有拼音的字符，详情请参考 :py:func:`~pypinyin.pinyin` :param strict: 是否严格遵照《汉语拼音方案》来处理声母和韵母，详见 :ref:`strict` :return: slug 字符串. 其中style定义了不同的效果： class Style(IntEnum): """拼音风格"""
    
        #: 普通风格，不带声调。如： 中国 -> ``zhong guo``
        NORMAL = 0
        #: 标准声调风格，拼音声调在韵母第一个字母上（默认风格）。如： 中国 -> ``zhōng guó``
        TONE = 1
        #: 声调风格2，即拼音声调在各个韵母之后，用数字 [1-4] 进行表示。如： 中国 -> ``zho1ng guo2``
        TONE2 = 2
        #: 声调风格3，即拼音声调在各个拼音之后，用数字 [1-4] 进行表示。如： 中国 -> ``zhong1 guo2``
        TONE3 = 8
        #: 声母风格，只返回各个拼音的声母部分（注：有的拼音没有声母，详见 `#27`_）。如： 中国 -> ``zh g``
        INITIALS = 3
        #: 首字母风格，只返回拼音的首字母部分。如： 中国 -> ``z g``
        FIRST_LETTER = 4
        #: 韵母风格，只返回各个拼音的韵母部分，不带声调。如： 中国 -> ``ong uo``
        FINALS = 5
        #: 标准韵母风格，带声调，声调在韵母第一个字母上。如：中国 -> ``ōng uó``
        FINALS_TONE = 6
        #: 韵母风格2，带声调，声调在各个韵母之后，用数字 [1-4] 进行表示。如： 中国 -> ``o1ng uo2``
        FINALS_TONE2 = 7
        #: 韵母风格3，带声调，声调在各个拼音之后，用数字 [1-4] 进行表示。如： 中国 -> ``ong1 uo2``
        FINALS_TONE3 = 9
        #: 注音风格，带声调，阴平（第一声）不标。如： 中国 -> ``ㄓㄨㄥ ㄍㄨㄛˊ``
        BOPOMOFO = 10
        #: 注音风格，仅首字母。如： 中国 -> ``ㄓ ㄍ``
        BOPOMOFO_FIRST = 11
        #: 汉语拼音与俄语字母对照风格，声调在各个拼音之后，用数字 [1-4] 进行表示。如： 中国 -> ``чжун1 го2``
        CYRILLIC = 12
        #: 汉语拼音与俄语字母对照风格，仅首字母。如： 中国 -> ``ч г``
        CYRILLIC_FIRST = 13

官方给的示例代码是这样的：

>>> import pypinyin
    >>> from pypinyin import Style
    >>> pypinyin.slug('中国人')
    'zhong-guo-ren'
    >>> pypinyin.slug('中国人', separator=' ')
    'zhong guo ren'
    >>> pypinyin.slug('中国人', style=Style.FIRST_LETTER)
    'z-g-r'
    >>> pypinyin.slug('中国人', style=Style.CYRILLIC)
    'чжун1-го2-жэнь2'

下面是使用sqlalchemy写的全部替换的方法，一般如果要更新的话，最好遍历出整个对象，之前因为考虑到要使用name，name\_acronym所以我就query这两个值加上id，后来发现赋值非常的麻烦。**所以更新最好遍历出要更新的某个值它所属的对象，这样可以直接对这个对象进行操作**。

迭代的时候报了个错，是因为遍历字典的时候没有获取每一项，这样字典就不是一个可迭代的对象，比如下面的迭代就是错误的：

for i, info in infos_dict:

报错为：

TypeError: cannot unpack non-iterable int object

完整的代码为：可以看到更新前1000个值耗时大概0.8秒,速度非常的慢

def get_quanpin_with_orm(session):
        result = session.query(Infos).filter(Infos.id <= 1000).all()
        infos_dict = { info.id: info for info in result}
        for i, info in infos_dict.items():
            info.name_acronym = slug(info.name, separator='', strict=False, errors='default')
        session.flush()
        session.commit()
    
    if __name__ == '__main__':
        time1 = time.process_time()
        session = DBSession()
        get_quanpin_with_orm(session)
        time2 = time.process_time()
        time = time2 - time1
        print("耗时————————————————", time)
       #耗时———————————————— 0.8202130000000003

下面我们使用sql来处理，因为要要处理赋值不一样的值，所以我们使用case when来更新，case when 的用法参考下面的博客：

[https://blog.csdn.net/NextAction/article/details/80453843?utm\_medium=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel\_param&depth\_1-utm\_source=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel\_param][https_blog.csdn.net_NextAction_article_details_80453843_utm_medium_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param_depth_1-utm_source_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param]

下面是使用sql更新的代码:

def get_quanpin_with_sql(session):
        sql1 = "select infos.id, infos.name from infos where infos.id <= 1000"
        result = session.execute(sql1)
        infos_name_acronym_dict = { }
        for i, name in result:
            infos_name_acronym_dict[i] = slug(name, separator='', strict=False, errors='default')
        sql2 = f"update infos set infos.name_acronym = case "
        for i, name_acronym in infos_name_acronym_dict.items():
            sql2 += f"when infos.id = {i} then '{name_acronym}' "
        sql2 += f"else infos.name_acronym end;"
        session.execute(sql2)
        session.flush()
        session.commit()
    
        #耗时———————————————— 0.05340999999999996

耗时0.05秒，比用sqlalchemy快了近16倍！！

在编写上述代码时也报了个错：

sqlalchemy.exc.ProgrammingError: (mysql.connector.errors.ProgrammingError) 1054 (42S22): Unknown column 'putao' in 'field list'

原因是在进行全拼的赋值时，sql语句变量的值应该用‘’单引号扩起来,因为定义的时候定义其为varchar类型。

sql2 += f"when infos.id = {i} then '{name_acronym}' "

以上实现了基础的功能，因为事实上数据量很大，一次性处理所有数据是不可能的，之前我试过一次处理，但是sql语句执行到一半就报错了，所以需要分也处理，每次处理1000条。

使用sql时，因为要执行两条sql语句，先查询需要的值，在更新数据库的值，所以我加入了id的判断，更新指定范围内的值。

def update_quanpin_with_sql_limit(session, id_start, id_end):
        id_from = int(id_start)
        id_end = int(id_end)
        while id_from <= id_end:
            id_to = id_from + 1000
            sql1 = f"select infos.id, infos.name from infos where infos.id between {id_from} and {id_to}"
            result = session.execute(sql1)
            infos_name_acronym_dict = { }
            for i, name in result:
                infos_name_acronym_dict[i] = slug(name, separator='', strict=False, errors='default')
            if infos_name_acronym_dict:
                sql2 = f"update infos set infos.name_acronym = case "
                for i, name_acronym in infos_name_acronym_dict.items():
                    sql2 += f"when infos.id = {i} then '{name_acronym}' "
                sql2 += f"end where infos.id between {id_from} and {id_to}"
                id_from += 1000
                process = id_from / id_end
                process = "%.2f%%" % (process * 100)
                print("当前进度为：", process)
                session.execute(sql2)
                session.flush()
                session.commit()
            else:
                id_from += 1000
                continue

**踩的坑：**

*  Sql1 查询和sql2 更新里面的id范围要一致，否则更新时将出现null值的情况，比如查询时是id\_from和id\_to,更新时id范围为id\_start和id\_end, 因为后者的范围较大，每次更新1000条，那么1000之外的数据就会为null
 *  使用case when更新时，then后面直接加上要赋值的内容即可，不用写成
    
    `then A.name_acronym = '{name-acronym}'`, 这样也是错误的，最终name\_acronym将被赋值为其他值
 *  因为表里的数据较多，甚至可能出现某个id范围内的数据没有，比如5000后的数据为8000，那么5000-8000这个范围的值就没有，那么拼接sql语句时就会发生错误，所以要加上判断，只有当查询出结果时才进行赋值，否则进行下一轮循环

最后测试本地线上数据的结果为：

![在这里插入图片描述][20200808113455882.png]  
可以看出更新的还是比较快的。

完整代码为：

def get_quanpin_with_orm(session):
    ‘’‘使用sqlalchemy集成的方法’‘’
        result = session.query(Infos).filter(Infos.id <= 1000).all()
        infos_dict = { info.id: info for info in result}
        for i, info in infos_dict.items():
            info.name_acronym = slug(info.name, separator='', strict=False, errors='default')
        session.flush()
        session.commit()
    
    def get_quanpin_with_sql(session):
      ‘’‘使用原生sql更新’‘’
        sql1 = "select infos.id, infos.name from infos where infos.id <= 1000"
        result = session.execute(sql1)
        infos_name_acronym_dict = { }
        for i, name in result:
            infos_name_acronym_dict[i] = slug(name, separator='', strict=False, errors='default')
        sql2 = f"update infos set infos.name_acronym = case "
        for i, name_acronym in infos_name_acronym_dict.items():
            sql2 += f"when infos.id = {i} then '{name_acronym}' "
        sql2 += f"else infos.name_acronym end;"
        session.execute(sql2)
        session.flush()
        session.commit()
    
    def update_quanpin_with_sql_limit(session, id_start, id_end):
      ‘’‘分页更新’‘’
        id_from = int(id_start)
        id_end = int(id_end)
        while id_from <= id_end:
            id_to = id_from + 1000
            sql1 = f"select infos.id, infos.name from infos where infos.id between {id_from} and {id_to}"
            result = session.execute(sql1)
            infos_name_acronym_dict = { }
            for i, name in result:
                infos_name_acronym_dict[i] = slug(name, separator='', strict=False, errors='default')
            if infos_name_acronym_dict:
                sql2 = f"update infos set infos.name_acronym = case "
                for i, name_acronym in infos_name_acronym_dict.items():
                    sql2 += f"when infos.id = {i} then '{name_acronym}' "
                sql2 += f"end where infos.id between {id_from} and {id_to}"
                id_from += 1000
                process = id_from / id_end
                process = "%.2f%%" % (process * 100)
                print("当前进度为：", process)
                session.execute(sql2)
                session.flush()
                session.commit()
            else:
                id_from += 1000
                continue
    
    if __name__ == '__main__':
        # time1 = time.process_time()
        # session = DBSession()
        # # get_quanpin_with_orm(session)
        # get_quanpin_with_sql(session)
        # time2 = time.process_time()
        # time = time2 - time1
        # print("耗时————————————————", time)
        time1 = time.process_time()
        id_start = input("请输入起始id：")
        id_end = input("请输入终止id：")
        session = DBSession()
        update_quanpin_with_sql_limit(session, id_start, id_end)
        time2 = time.process_time()
        time = time2 - time1
        print("耗时————————————————————————", time)

[https_blog.csdn.net_NextAction_article_details_80453843_utm_medium_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param_depth_1-utm_source_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param]: https://blog.csdn.net/NextAction/article/details/80453843?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param
[20200808113455882.png]: /images/20221124/5e0629b1478b4271b20679e6513a6942.png