[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

原文[https://www.cnblogs.com/xiaomengceshi/p/13663964.html](https://www.cnblogs.com/xiaomengceshi/p/13663964.html)





1、从一个序列中随机选取一个元素返回：

  random.choice(sep)

2、用于将一个列表中的元素打乱

  random.shuffle(sep)

 3、在sep列表中随机选取k个元素返回

   random.sample(seq, k) 

4、随机返回a到b之间的一个整数

 　 random.randint(a,b)

5、随机返回0到1之间的浮点数

  random.random()

6、从指定范围内，按指定基数递增的集合中 获取一个随机数

  random.randrange([start], stop[, step]) 

 7、随机返回a到b之间的一个浮点数

random.uniform(a, b)

python中生成随机整数（random模块）

## 基本操作





Series是一列，一般情况下别用，尽量转到DataFrame.
```python
import numpy as np
import pandas as pd
s = pd.Series([1, 3, 5, np.nan, 6, 8])
print(s)
s = pd.Series([1, 3, 5, None, 6, 8])
print(s)
```

选取Series中的一个数据，[]里面的是Series的index，如果idnex不对下面这句话就会报错：

```python
s[1]
```

```python
s = pd.Series([1, 3, 5, np.nan, 6, 8])
print(s.values)
print(s.to_dict())
```

Series转DataFrame：
```python
s.to_frame()
```

DataFrame转Series，不用转，直接选取某一列就是Series，下面的df3就是Series：
```python
df3=df['A']
```

DataFrame

读取数据：

```python
import numpy as np
import pandas as pd
df = pd.DataFrame({'A': 1.,
                   'B': pd.Timestamp('20130102'),
                   'C': pd.Series(1, index=list(range(4)), dtype='float32'),
                   'D': np.array([3] * 4, dtype='int32'),
                   'E': pd.Categorical(["test", "train", "test", "train"]),
                   'F': 'foo',
                   'G': [pd.Timestamp('2021-06-28 01:00:00'), pd.Timestamp('2021-06-28 02:00:00'),
                         pd.Timestamp('2021-06-28 03:00:00'), pd.Timestamp('2021-06-28 04:00:00'), ],
                   'H': [2, 30, 23, 12],
                   'I': ["123", "98", "23", "2"],
                   'X': [1, 2, None, np.nan]})
```

DataFrame广播性质

```python
df['G'] = 123  # 新增一列
df['A'] = 456  # 原来同名列被替代
print(df)
```

给某一列一个列表，列表长度要足够分配。
```python
df['A'] = [2,3,4,5] 
```

查看所有列是什么对象：
```python
df.dtypes
```

得到所有的列名：

```python
list(df.columns)
```

选择某些子列组成新的DataFrame，不用copy就容易引起警告，df=df[['B','C','A']]这样写注销掉之前的df也是可以无警告的：
```python
df2=df[['B','C','A']].copy()
```

更新所有的列名，长度需要相等,下面长度会报错：
```python
df2.columns=["TIME",'fl','vl']
```

根据列名删除某一列：

```python
df2.pop('C')
```

选取某一行某一列的某个数值，iloc重载了[]运算符，直接选第0行第1列：

```python
df.iloc[0,1]
```

iloc不便于给人看，最好是用loc：

```python
df.loc[0,'B']
```

此时赋值也是可以的：

```python
df.loc[0,'A']=10
```

loc还有更大的作用，下面2句话等价，都是选取满足条件的行：

```python
df.loc[df['E']=='test']
df.loc[df['E']=='test',:]
```

选取满足多个条件的那些行，注意区别&运算 和 | 运算 ：

```python
df.loc[(df['E']=='test')&(df['A']==1)]
df.loc[(df['E']=='test')|(df['A']==1)]
```

当然，可以进一步选择，在满足条件的一行里面进一步选某一列或者多列：

```python
df.loc[df['E']=='test','A']  # 结果是一个series
```

多个条件取值：
```python
df.loc[(df['E']=='test')&(df['A']==1),'B'].values[0] # 条件筛选-->某一列-->第一个数值
```

遍历每一行，如果是要依靠for循环去修改每一行的数值，那最好用apply而不是这种方式：

```python
for index, row in df.iterrows():
    print(index) # 输出每行的索引值
```

重置index
```python
df.reset_index(drop=True)   # drop=True 会丢弃之前的index
```

某一列的元素组成一个集合：

```python
df['A'].unique()  # 某一列不重复的数值
```




## 中级操作


将DataFrame变回python数据结构，我最常用参数是records，配合函数方便我往数据库写数据：
```python
df.to_dict(orient='records')
```

![在这里插入图片描述](/static/img/images/00826.png)

根据某一列的数值生成另一列：

```python
df['D']=df['G'].apply(lambda x:str(x)[0:10])
```

根据某两列的数值生成另一列，根据某几列的数值生成另一列同理：

```python
df['E']=df[['C','A']].apply(lambda x:x['A']+x['C'],axis=1)
```

df['D']此时是字符串，将字符串转换为日期：

```python
df['D']=pd.to_datetime(df['D'])
```

删除某一行，传入行index：

```python
df=df.drop(0)
```

删除数值满足条件的行
```python
df = df.drop(df[df['B']>2.6].index)
df = df.drop(df[df['B'].isin([np.nan])].index)
```

删除所有含有空值的行，any表示某一行有空值就会删除整行，all表示整行都是空值才删除整行：
```python
df.dropna(axis=0, how='any', inplace=True)
df=df.dropna(axis=0, how='any')
```

删除所有含有空值的列，any表示某一列有空值就会删除整列，all表示整列都是空值才删除整列：
```python
df.dropna(axis=1, how='any', inplace=True)
df=df.dropna(axis=1, how='any')
```

整张表都有哪些是空值，哪些不是：
```python
df.isnull()
```

判断表是不是存在空值，哪怕有一个空值都会返回True：
```python
np.any(df.isnull())
```

判断表是不是每个元素都是空值：
```python
np.all(df.isnull())
```

填充空值，如何填充空值：

将所有空值填充为0：

```python
df.fillna(0)
```

在列里面，从后面向前面填充，最后一行空值还是会是空：
```python
df.fillna(method=‘bfill’)
```

在列里面，从前面向后面填充，第一行空值还是会是空：
```python
df.fillna(method=‘ffill’)
```

横向填充，从前面向后面填充，第一列空值还是会是空：
```python
df.fillna(method=‘ffill’,axis=1)
```

处理空值的另一种方式，插值，线性插值，前向，竖着填充。

```python
df.interpolate(method=‘linear’, limit_direction=‘forward’, axis=0)
```



## 高级操作


pandas 表里所有负数赋为空值，消除所有负数：

[https://blog.csdn.net/x1131230123/article/details/117003108](https://blog.csdn.net/x1131230123/article/details/117003108)


关联性分析,相关性系数计算：
```python
data.corr()  #相关系数矩阵，即给出了两个物品之间的相关系数
data.corr()[u'A物品']  #只显示A物品与其他菜式的相关系数
data[u'A物品'].corr(data[u'B物品'])  #计算A物品与B物品的相关系数
```



print('series通过索引进行排序：'

print(series.sort_index())

print('series通过值进行排序:'

print(series.sort_values())

print('dataframe根据行索引进行降序排序（排序时默认升序，调节ascending参数）)：'

print(frame.sort_index(ascending=False))

print('dataframe根据列索引进行排序：')

print(frame.sort_index(axis=1))

print('dataframe根据值进行排序：')

print(frame.sort_values(by='a'))

print('通过多个索引进行排序：')

print(frame.sort_values(by=['a','c']))


2列转换成字典关系：

```python
bind_adcode = city_df[['bind', 'adcode']].set_index('bind').to_dict()['adcode']
```

去重

```
df= df.drop_duplicates(subset="device_no", keep='first', inplace=False)
```


## 进一步学习

学习了基本操作后，进一步巩固就多看官网api：

[https://pandas.pydata.org/docs/reference/io.html](https://pandas.pydata.org/docs/reference/io.html)

pandas cook book myself 白话文教程

[https://www.cnblogs.com/haoabcd2010/p/10614509.html](https://www.cnblogs.com/haoabcd2010/p/10614509.html)




```python
import time
from concurrent import futures
from concurrent.futures import ThreadPoolExecutor
def display(args):
    print(time.strftime('[%H:%M:%S]', time.localtime()), end=' ')
    print(args)
def task(n):
    """只是休眠"""
    display('begin sleep {}s.'.format(n))
    time.sleep(n)
    display('ended sleep {}s.'.format(n))
def do_many_task_inorder():
    """多线程
    按任务发布顺序依次等待完成
    """
    tasks = [5, 4, 3, 2, 1]
    with ThreadPoolExecutor(max_workers=3) as executor:
        future_list = [executor.submit(task, arg) for arg in tasks]
        display('非阻塞运行')
        for future in future_list:
            display(future)
        display('统一结束(有序)')
        for future in future_list:
            display(future.result())
def do_many_task_disorder():
    """多线程执行
    先完成先显示
    """
    tasks = [5, 4, 3, 2, 1]
    with ThreadPoolExecutor(max_workers=3) as executor:
        future_list = [executor.submit(task, arg) for arg in tasks]
        display('非阻塞运行')
        for future in future_list:
            display(future)
        display('统一结束(无序)')
        done_iter = futures.as_completed(future_list)  # generator
        for done in done_iter:
            display(done)
if __name__ == '__main__':
    do_many_task_inorder()
    do_many_task_disorder()
```

python 线程池 ThreadPoolExecutor 用法

https://blog.csdn.net/lwgkzl/article/details/82145387





split()不带参数即可
```python
def test_filter():
    str = "aa     bbbbb         ccc  d"
    str_list = str.split()
    print str_list
```

数据之间空格不一样 python 两种方法分割python多空格字符串

`__call__()`方法作用在于：





类-->实例化-->实例也能当成一个可调用对象

net = Net() #类的实例

out = net(input)  #实例也能当成一个可调用对象

```python
import torch
import torch.nn as nn
import torch.nn.functional as F
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        # 输入图像channel：1；输出channel：6；5x5卷积核
        self.conv1 = nn.Conv2d(1, 6, 5)
        self.conv2 = nn.Conv2d(6, 16, 5)
        # an affine operation: y = Wx + b
        self.fc1 = nn.Linear(16 * 5 * 5, 120)
        self.fc2 = nn.Linear(120, 84)
        self.fc3 = nn.Linear(84, 10)
    def forward(self, x):
        # 2x2 Max pooling
        x = F.max_pool2d(F.relu(self.conv1(x)), (2, 2))
        # 如果是方阵,则可以只使用一个数字进行定义
        x = F.max_pool2d(F.relu(self.conv2(x)), 2)
        x = x.view(-1, self.num_flat_features(x))
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        x = self.fc3(x)
        return x
    def num_flat_features(self, x):
        size = x.size()[1:]  # 除去批处理维度的其他所有维度
        num_features = 1
        for s in size:
            num_features *= s
        return num_features
net = Net()
print(net)
params = list(net.parameters())
print(len(params))
print(params[0].size())  # conv1's .weight
input = torch.randn(1, 1, 32, 32)
out = net(input)
print(out)
```

文章：

[http://c.biancheng.net/view/2380.html](http://c.biancheng.net/view/2380.html)

```python
class CLanguage:
    # 定义__call__方法
    def __call__(self,name,add):
        print("调用__call__()方法",name,add)
clangs = CLanguage()
clangs("C语言中文网","http://c.biancheng.net")
```

> Python 中，凡是可以将 () 直接应用到自身并执行，都称为可调用对象。可调用对象包括自定义的函数、Python 内置函数以及本节所讲的类实例对象。


文章：

[https://blog.csdn.net/Yaokai_AssultMaster/article/details/70256621](https://blog.csdn.net/Yaokai_AssultMaster/article/details/70256621)

Python __call__()方法 __init__()方法

# Pandas DataFrame to List of Dictionaries:





[https://stackoverflow.com/questions/29815129/pandas-dataframe-to-list-of-dictionaries/29816143#29816143](https://stackoverflow.com/questions/29815129/pandas-dataframe-to-list-of-dictionaries/29816143#29816143)

![在这里插入图片描述](/static/img/images/00849.png)

```python
df.to_dict('records')
```

# Convert list of dictionaries to a pandas DataFrame:

[https://stackoverflow.com/questions/20638006/convert-list-of-dictionaries-to-a-pandas-dataframe](https://stackoverflow.com/questions/20638006/convert-list-of-dictionaries-to-a-pandas-dataframe)

![在这里插入图片描述](/static/img/images/00850.png)

![在这里插入图片描述](/static/img/images/00851.png)


[https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_sql.html#pandas.read_sql](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_sql.html#pandas.read_sql)

[https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_sql_table.html#pandas.read_sql_table](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_sql_table.html#pandas.read_sql_table)