Python





```python
from collections import defaultdict
def with_running_count(values):
    counts = defaultdict(int)
    for value in values:
        counts[value] += 1
        yield value, counts[value]
print(*with_running_count('abaab'))
```

yield 生成器 生成 *

[https://docs.python.org/3/library/multiprocessing.html](https://docs.python.org/3/library/multiprocessing.html)





# 1 背景

python由于[Global Interpreter Lock](https://docs.python.org/3/glossary.html#term-global-interpreter-lock) 的原因，多线程并不是我们想的那种CPU调度的多线程，在python中开了多线程其实也是一个单进程。

针对CPU密集型（CPU-bound）的任务，比如计算气象数据，多线程和多进程调用多个CPU核心并行，能有效减少任务计算时间，提高实时性，但python的多线程是不ok的。

针对IO密集型（I/O bound）的任务，比如爬虫，多线程和多进程也是OK的，python的多线程显然也是ok的。

进程会占用更多资源，比如主进程开一个子进程，子进程会把主进程的变量啥的复制一份，这也导致进程之间通信和线程不一样。多线程的时候，给一个全局列表进去还能线程安全；多进程的时候，子进程只会把主进程的全局列表复制一份带入，修改的也是修改子进程的。所以需要类似信号、管道、消息队列、共享内存的东西来进行，python multiprocessing有消息队列、管道、共享内存，优点是快，缺点是不太灵活（存储类型和运行机制），python multiprocessing提供了一个更高级的封装类`Manager`来专门处理这个事情，`Manager`自己就是一个服务进程，有点就是灵活，缺点就是慢一些。


# 2 进程池

进程池肯定是很爽的，使用简单，效果显著。

windows需要multiprocessing.freeze_support()

Linux不需要。


```python
import multiprocessing
import time
import os
def func1(inputdata, res):
    res.append([sum(inputdata)])
    print("res此时的结果： ", res)
    print("此时的时间： ", time.time())
    print("进城的PID： ", os.getpid())
if __name__ == '__main__':
    multiprocessing.freeze_support()
    res = multiprocessing.Manager().list()  # 通信
    inputdata = [i for i in range(100)]  # 随便弄点数
    p = multiprocessing.Pool(2)  # 创建一个包含2个进程的进程池
    p.apply_async(func=func1, args=(inputdata[0:10], res,))  # 往池子里加一个异步执行的子进城
    p.apply_async(func=func1, args=(inputdata[10:20], res,))  # 往池子里加一个异步执行的子进程
    p.apply_async(func=func1, args=(inputdata[20:30], res,))  # 往池子里加一个异步执行的子进程
    p.close()  # 等子进程执行完毕后关闭进程池
    p.join()  # 主进程等待
```

运行结果：

```python
res此时的结果：  [[45]]
此时的时间：  1610254954.318595
进城的PID：  15372
res此时的结果：  [[45], [245]]
此时的时间：  1610254954.3205965
进城的PID：  15372
res此时的结果：  [[45], [245], [145]]
此时的时间：  1610254954.321597
进城的PID：  16268
```

multiprocessing — Process-based parallelism python 多进程 用法 进程池

# pymysql封装





只封装了select选择数据和insert插入数据：
```python
# encoding: utf-8
import traceback
import pymysql
import pandas as pd
class easy_pymysql():
    # 初始化
    def __init__(self, host, user, passwd):
        self.host = host
        self.user = user
        self.passwd = passwd
        self.conn = None
        self.cursor = None
    def connect(self):
        self.conn = pymysql.connect(host=self.host,
                                    user=self.user,
                                    password=self.passwd,
                                    # database=self.dbName,
                                    use_unicode=True,
                                    charset='utf8')
        self.cursor = self.conn.cursor()
    def close(self):
        self.cursor.close()
        self.conn.close()
    def select_data(self, sql):
        result = None
        self.connect()
        try:
            self.cursor.execute(sql)
            result = self.cursor.fetchall()
            result = pd.DataFrame(result)
        except:
            traceback.print_exc()
            self.conn.rollback()
        self.close()
        return result
    def insert(self, df: pd.DataFrame, tablename: str):
        df = df.to_dict(orient="split")
        tup_col = ", ".join(df['columns'])
        datas = str([tuple(i) for i in df['data']])[1:-1]
        sql = f"INSERT INTO {tablename} ({tup_col}) VALUES {datas}"
        self.connect()
        try:
            self.cursor.execute(sql)
            self.conn.commit()
        except:
            traceback.print_exc()
            self.conn.rollback()
        self.close()
    def insert_table(self, df: pd.DataFrame, tablename: str):
        if len(df) <= 20000:
            self.insert(df, tablename)
        else:
            n = len(df) // 20000
            for i in range(n):
                df_sub = df[20000 * i:(20000 * (i + 1))]
                self.insert(df_sub, tablename)
            df_sub = df[(20000 * (i + 1)):]
            self.insert(df_sub, tablename)
        print(f"finish transfer mysql {len(df)}")
    def update_table(self, df: pd.DataFrame, tablename: str):
        self.connect()
        try:
            idchname, upchname = list(df.columns)
            print(idchname, upchname)
            id_list = list(df[idchname])
            upch_list = list(df[upchname])
            for id_, upch_ in zip(id_list, upch_list):
                sql = f"UPDATE {tablename} SET {upchname}={upch_} WHERE {idchname}={id_}"
                self.cursor.execute(sql)
            self.conn.commit()
        except:
            traceback.print_exc()
            self.conn.rollback()
        self.close()
```

# clickhouse_driver封装

只封装了select选择数据和insert插入数据：

```python
# encoding: utf-8
from clickhouse_driver.client import Client
import pandas as pd
class easy_clickhouse():
    # 初始化
    # def __init__(self, host, user="", passwd=""):
    def __init__(self, host):
        self.host = host
        self.client = Client(host=self.host)
    def select_data(self, sql):
        result = self.client.execute(sql)
        if len(result) == 0:
            return None
        else:
            result = pd.DataFrame(result)
        return result
    def insert(self, df: pd.DataFrame, tablename: str):
        df = df.to_dict(orient="split")
        tup_col = ", ".join(df['columns'])
        datas = str([tuple(i) for i in df['data']])[1:-1]
        sql = f"INSERT INTO {tablename} ({tup_col}) VALUES {datas}"
        self.client.execute(sql)
    def insert_table(self, df: pd.DataFrame, tablename: str):
        if len(df) <= 20000:
            self.insert(df, tablename)
        else:
            n = len(df) // 20000
            for i in range(n):
                df_sub = df[20000 * i:(20000 * (i + 1))]
                self.insert(df_sub, tablename)
            df_sub = df[(20000 * (i + 1)):]
            self.insert(df_sub, tablename)
        print(f"finish transfer clickhouse {len(df)}")
```

```python
def insert_db_table(datadict_list):
    '''
    datadict_list是字典列表  每个字典是一条记录
    开发环境的
    '''
    host = '10.10.90.11'
    table = "tbs"
    sql = "insert into %s " % (table)
    c = clientclickhouse(host=host)
    sum1 = 0
    for datadict in datadict_list:
        result = datadict  # 字典
        names = ["%s" % x for x in result.keys()]
        names = ",".join(names)
        if sum1 == 0:
            sql = sql + "(%s) values " % (names)
        values = ["'%s'" % x if isinstance(x, str) else "{}".format(x) for x in result.values()]
        values = ",".join(values)
        sql = sql + "(%s)," % (values)
        sum1 = sum1 + 1
    sql_str = sql[0:-1]
    c.execute(sql_str)
```

python pymysql 插入多条数据 字典列表 数据 clickhouse driver 封装 SQL

原文：





[https://blog.csdn.net/dl1456074580/article/details/86538098](https://blog.csdn.net/dl1456074580/article/details/86538098)

```python
def skip_ws(txt, pos):
    while pos < len(txt) and txt[pos].isspace():
        pos += 1
    return pos
def parse_str(txt, pos, allow_ws=False, delimiter=[',', ':', '}', ']']):
    while pos < len(txt):
        if not allow_ws and txt[pos].isspace():
            break
        if txt[pos] in delimiter:
            break
        pos += 1
    return pos
def parse_obj(txt, pos):
    obj = dict()
    while True:
        pos = skip_ws(txt, pos + 1)
        end = parse_str(txt, pos, True, [':'])
        if end >= len(txt):
            raise ValueError("unexpected end when parsing object key")
        key = txt[pos:end].strip()
        pos = skip_ws(txt, end + 1)
        if pos >= len(txt):
            raise ValueError("unexpected end when parsing object value")
        if txt[pos] == '[':
            value, pos = parse_array(txt, pos)
        elif txt[pos] == '{':
            value, pos = parse_obj(txt, pos)
        else:
            end = parse_str(txt, pos, True, [',', '}'])
            if end >= len(txt):
                raise ValueError("unexpected end when parsing object value")
            value = txt[pos:end].strip()
            pos = end
        obj[key] = value
        pos = skip_ws(txt, pos)
        if pos >= len(txt):
            raise ValueError("unexpected end when object value finish")
        if txt[pos] == '}':
            return obj, pos + 1
def parse_array(txt, pos):
    array = list()
    while True:
        pos = skip_ws(txt, pos + 1)
        if pos >= len(txt):
            raise ValueError("unexpected end when parsing array item")
        if txt[pos] == '[':
            value, pos = parse_array(txt, pos)
        elif txt[pos] == '{':
            value, pos = parse_obj(txt, pos)
        else:
            end = parse_str(txt, pos, True, [',', ']'])
            if end >= len(txt):
                raise ValueError("unexpected end when parsing array item")
            value = txt[pos:end].strip()
            pos = end
        array.append(value)
        pos = skip_ws(txt, pos)
        if pos >= len(txt):
            raise ValueError("unexpected end when array item finish")
        if txt[pos] == ']':
            return array, pos + 1
def parse(txt):
    if txt.startswith('json'):
        pos = txt.find(':')
        if pos != -1:
            pos = skip_ws(txt, pos + 1)
            if txt[pos] == '{':
                obj, pos = parse_obj(txt, pos)
                return obj
            elif txt[pos] == '[':
                array, pos = parse_array(txt, pos)
                return array
    raise ValueError("format error when parsing root")
if __name__ == '__main__':
    import json
    txt = r'json:{name:张三,age:22,book:[{bookname:语文,booksize:200,soce:100},{wkke:123}]}'
    resdict = parse(txt)
    print(resdict)
    print(type(resdict))
    restxt=json.dumps(resdict,ensure_ascii=False)
    print(restxt)
    print(type(restxt))
    resd=json.loads(restxt)
    print(resd)
    print(type(resd))
```

Python转类json格式的数据（没有双引号的JSON格式，一般在硬件对接中出现）

创建一个




```c
conda config --add channels r
```

conda 找不到 .condarc