python 窗口分发

python

字数统计: 605阅读时长: 3 min

 2018/05/20   Share

对于时序数据的处理，会涉及到window窗口的概念，就比如每几秒(秒)返回前几秒(window)处理的的result，就如每1分钟返回这一分钟nginx日志的请求url的热区数据，为了学习这些，我自己先写个简单的用例，一步步加深。

返回每5秒前10秒钟产生数据的平均值

import datetime
import random
import time


def window(source ,handler, interval: int,width: int):
    store = []
    startime = datetime.datetime.now()
    while True:
        data = next(source)   
        if data:
            store.append(data)
            currentime = data['time']
        if (currentime - startime).total_seconds() >= interval:   #对比时间间隔是否符合interval
            startime = currentime
            dt = currentime - datetime.timedelta(seconds=width)     # 设置窗口大小
            store = [x  for x in store if x['time'] > dt]           #保留下符合窗口大小的数据,width >= interval
            handler(store)


def source():
    while True
        data = {'time':datetime.datetime.now(),'value': random.randint(0,100)}
        yield data
        time.sleep(1)


def handler(items):
    values = [item['value'] for item in items]
    print('随机数平均值:',round(sum(values)/len(values),2))

def main(interval,width):
    data = source()
    window(data,handler,interval,width)

if __name__ == "__main__":
    main(5,10)

以上的代码我们只有一个处理进程在工作，如果我们需要进行多处理进程工作，我们就需要涉及分发的概念。

从单一的 source -> worker 到 source -> 分发器 -> 多个worker

分发器实践

import datetime
import random
import time
import queue
import multiprocessing
from functools import wraps

# 进程间队列
analyers = queue.Queue()

# 数据生产
def source():
    while True:
        data = {'time':datetime.datetime.now(),'value': random.randint(0,100)}
        yield data
        time.sleep(1)

data = source()

# 进程队列装饰器
def dispatcher(data,interval,width):
    def register(fn):
        @wraps(fn)
        def wrap(*args,**kwargs):
            p = multiprocessing.Process(target=window,args=(data,fn,interval,width))
            analyers.put(p)
        return wrap
    return register

# 队列获取，子进程启动
def start():
    for _ in range(analyers.qsize()):
        analyers.get_nowait().start()


# 窗口函数
def window(source ,handler, interval: int,width: int):
    store = []
    startime = datetime.datetime.now()
    while True:
        data = next(source)   
        if data:
            store.append(data)
            currentime = data['time']
        if (currentime - startime).total_seconds() >= interval:   #对比时间间隔是否符合interval
            startime = currentime
            dt = currentime - datetime.timedelta(seconds=width)  
            store = [x  for x in store if x['time'] > dt]           #保留下符合窗口大小的数据,width >= interval
            handler(store)


@dispatcher(data,5,10)
def handler(*args,**kwargs):
    values = [item['value'] for item in args[0]]
    print('随机数平均值:',round(sum(values)/len(values),2))


@dispatcher(data,5,10)
def handler1(*args,**kwargs):
    values = [item['value'] for item in args[0]]
    print('随机数总和:',sum(values))

if __name__ == "__main__":
    handler()
    handler1()
    start()

原文作者：Zhang Jusene

原文链接：http://jusene.github.io/2018/05/20/window/

发表日期：May 20th 2018, 10:12:38 pm

更新日期：November 30th 2019, 1:54:24 am

Next Post

watchdog 监控文件系统事件
Previous Post

IO网络模型

CATALOG



缺失模块。
1、请确保node版本大于6.2
2、在博客根目录（注意不是archer根目录）执行以下命令：
npm i hexo-generator-json-content --save
3、在根目录_config.yml里添加配置：

jsonContent:
  meta: false
  pages: false
  posts:
    title: true
    date: true
    path: true
    text: false
    raw: false
    content: false
    slug: false
    updated: false
    comments: false
    link: false
    permalink: false
    excerpt: false
    categories: true
    tags: true

1	import datetime
2	import random
3	import time
4
5
6	def window(source ,handler, interval: int,width: int):
7	store = []
8	startime = datetime.datetime.now()
9	while True:
10	data = next(source)
11	if data:
12	store.append(data)
13	currentime = data['time']
14	if (currentime - startime).total_seconds() >= interval: #对比时间间隔是否符合interval
15	startime = currentime
16	dt = currentime - datetime.timedelta(seconds=width) # 设置窗口大小
17	store = [x for x in store if x['time'] > dt] #保留下符合窗口大小的数据,width >= interval
18	handler(store)
19
20
21	def source():
22	while True
23	data = {'time':datetime.datetime.now(),'value': random.randint(0,100)}
24	yield data
25	time.sleep(1)
26
27
28	def handler(items):
29	values = [item['value'] for item in items]
30	print('随机数平均值:',round(sum(values)/len(values),2))
31
32	def main(interval,width):
33	data = source()
34	window(data,handler,interval,width)
35
36	if __name__ == "__main__":
37	main(5,10)