"""鑒于項(xiàng)目保密需要，以下代碼僅為示例"""
import time
import pathlib as pl
import pandas as pd
from threading import Thread
from multiprocessing import Queue, Process, cpu_count
# 導(dǎo)入多線程Thread,多進(jìn)程的隊(duì)列Queue,多進(jìn)程Process，CPU核數(shù)cpu_count
# 存放分段讀取的數(shù)據(jù)隊(duì)列，注：maxsize控制隊(duì)列的最大數(shù)量，避免一次性讀取到內(nèi)存中的數(shù)據(jù)量太大
data_queue = Queue(maxsize=cpu_count() * 2)  
# 存放等待寫入磁盤的數(shù)據(jù)隊(duì)列
write_queue = Queue()  
def read_data(path: pl.Path, data_queue: Queue, size: int = 10000):
    """
    讀取數(shù)據(jù)放入隊(duì)列的方法
    :return:
    """
    data_obj = pd.read_csv(path, sep=',', header=0, chunksize=size, dtype='category')
    for idx, df in enumerate(data_obj):
        while data_queue.full():  # 如果隊(duì)列滿了，那就等待
            time.sleep(1)
        data_queue.put((idx + 1, df))
    data_queue.put((None, None))  # 放入結(jié)束信號(hào)
def write_data(out_path: pl.Path, write_queue: Queue):
    """
    將數(shù)據(jù)增量寫入CSV的方法
    :return:
    """
    while True:
        while write_queue.empty():
            time.sleep(1)
        idx, df = write_queue.get()
        if df is None:
            return  # 結(jié)束退出
        df.to_csv(out_path, mode='a', header=None, index=False, encoding='ansi')  # 輸出CSV
def parse_data(data_queue: Queue, write_queue: Queue):
    """
    從隊(duì)列中取出數(shù)據(jù)，并加工的方法
    :return:
    """
    while True:
        while write_queue.empty():
            time.sleep(1)
        idx, df = data_queue.get()
        if df is None:  # 如果是空的結(jié)束信號(hào)，則結(jié)束退出進(jìn)程，
        # 特別注意結(jié)束前把結(jié)束信號(hào)放回隊(duì)列，以便其他進(jìn)程也能接收到結(jié)束信號(hào)！??！
            data_queue.put((idx, df))
            return
        """處理數(shù)據(jù)的業(yè)務(wù)邏輯略過"""
        write_queue.put((idx, df))  # 將處理后的數(shù)據(jù)放入寫隊(duì)列
# 創(chuàng)建一個(gè)讀取數(shù)據(jù)的線程
read_pool = Thread(target=read_data, args=(read_data_queue, *args))
read_pool.start()  # 開啟讀取線程
# 創(chuàng)建一個(gè)增量寫入CSV數(shù)據(jù)的線程
write_pool = Thread(target=write_data, args=(write_data_queue, *args))
write_pool.start()  # 開啟寫進(jìn)程
pools = []  # 存放解析進(jìn)程的隊(duì)列
for i in range(cpu_count()):  # 循環(huán)開啟多進(jìn)程，不確定開多少個(gè)進(jìn)程合適的情況下，那么按CPU的核數(shù)開比較合理
    pool = Process(target=parse_data, args=(read_data_queue, write_data_queue, *args))
    pool.start()  # 啟動(dòng)進(jìn)程
    pools.append(pool)  # 加入隊(duì)列
for pool in pools:
    pool.join()  # 等待所有解析進(jìn)程完成
# 所有解析進(jìn)程完成后，在寫隊(duì)列放入結(jié)束寫線程的信號(hào)
write_data_queue.put((None, None))  
write_pool.join()  # 等待寫線程結(jié)束
print('任務(wù)完成')

"""鑒于保密需要，以下代碼僅為示例"""
class Obj:
    def __init__(self, ser: pd.Series):
        """
        初始化類對(duì)象
        :param ser: 傳入series
        """
        self.ser = ser  # 行數(shù)據(jù)
        self.attrs1 = []  # 屬性1
        self.attrs2 = []  # 屬性2
        self.attrs3 = []  # 屬性3
    def __repr__(self):
        """
        自定義輸出
        """
        attrs1 = '_'.join([str(a) for a in self.attrs1])
        attrs2 = '_'.join([str(a) for a in self.attrs2])
        attrs3 = '_'.join([str(a) for a in self.attrs3])
        return '_'.join([attrs1, attrs2, attrs3])
    def run(self):
        """運(yùn)行業(yè)務(wù)邏輯"""
# 創(chuàng)建obj列，存入類對(duì)象
data['obj'] = data.apply(lambda x: Obj(x), axis=1)
# 運(yùn)行obj列中的類方法獲得判斷結(jié)果
data['obj'] = data['obj'].map(lambda x: x.run())
# 鏈?zhǔn)秸{(diào)用，1將類對(duì)象文本化->2拆分到多列->3刪除空列->4轉(zhuǎn)換為category格式
data[['col1', 'col2', 'col3', ...省略]] = data['obj'].map(str).str.split('_', expand=True).dropna(axis=1).astype('category')
# 刪除obj列
data.drop(columns='obj', inplace=True)  

"""numpy方法使用示例"""
import numpy as np
# 計(jì)算數(shù)字的個(gè)數(shù)組合bincount
np.bincount([9, 2, 13, 12, 9, 10, 11])
# 輸出結(jié)果：array([0, 0, 1, 0, 0, 0, 0, 0, 0, 2, 1, 1, 1, 1], dtype=int64)
# 取得個(gè)數(shù)最多的數(shù)字argmax
np.argmax(np.bincount([9, 2, 13, 12, 9, 10, 11]))
# 輸出結(jié)果: 9
# 將數(shù)字按照個(gè)數(shù)優(yōu)先，其次大小進(jìn)行排序argsort
np.argsort(np.bincount([9, 2, 13, 12, 9, 10, 11]))
# 輸出結(jié)果：array([ 0,  1,  3,  4,  5,  6,  7,  8,  2, 10, 11, 12, 13,  9], dtype=int64)
# 翻轉(zhuǎn)列表flipud
np.flipud(np.argsort(np.bincount([9, 2, 13, 12, 9, 10, 11])))
# 輸出結(jié)果: array([ 9, 13, 12, 11, 10,  2,  8,  7,  6,  5,  4,  3,  1,  0], dtype=int64)
# 查找相同值in1d
np.in1d([2, 3, 4], [2, 9, 3])
# 輸出結(jié)果: array([ True,  True, False]) 注：指2,3True，4False
np.all(np.in1d([2, 3], [2, 9, 3]))
# 輸出結(jié)果: array([ True,  True])
# 是否全是all
np.all(np.in1d([2, 3, 4], [2, 9, 3]))  # 判斷組合1是否包含在組合2中
# 輸出結(jié)果: False
np.all(np.in1d([2, 3], [2, 9, 3]))
# 輸出結(jié)果: True