【GitHub】Pyleus

相似文件 换一批

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

Pyleus

Pyleus is a Python 2.6+ framework for developing and launching Apache Storm topologies.

Please visit our documentation.

masterdevelop
About

Pyleus is a framework for building Apache Storm topologies in idiomatic Python.

With Pyleus you can:

• define a topology with a simple YAML file
• have dependency management with a requirements.txt file
• run faster thanks to Pyleus’ MessagePack based serializer
• pass options to your components directly from the YAML file
• use the Kafka spout built into Storm with only a YAML change
  

本帖隐藏的内容

pyleus-develop.zip (123.7 KB)

2017-3-24 08:28:25 上传

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：GitHub Hub Requirements Requirement Developing management framework building directly simple

1. from __future__ import absolute_import
   
2. 
   
3. from array import array
   
4. from collections import defaultdict
   
5. import logging
   
6. import time
   
7. 
   
8. from pyleus.storm import SimpleBolt
   
9. 
   
10. from apparent_temperature.measure_generator import Measure
    
11. 
    
12. log = logging.getLogger('heat_index_bolt')
    
13. 
    
14. CX = (
    
15.     -42.379,
    
16.     2.04901523,
    
17.     10.14333127,
    
18.     -0.22475541,
    
19.     -6.83783 * 0.001,
    
20.     -5.481717 * 0.01,
    
21.     1.22874 * 0.001,
    
22.     8.5282 * pow(10, -4),
    
23.     -1.99 * pow(10, -6))
    
24. 
    
25. 
    
26. def _heat_index(temp, hum):
    
27.     temp2 = temp * temp
    
28.     hum2 = hum * hum
    
29.     factors = (1, temp, hum, temp * hum, temp2, hum2, temp2 * hum, temp * hum2,
    
30.                temp2 * hum2)
    
31.     heat = 0
    
32.     for idx in range(len(CX)):
    
33.         heat += CX[idx] * factors[idx]
    
34.     return round(heat, 2)
    
35. 
    
36. 
    
37. class HeatIndexBolt(SimpleBolt):
    
38. 
    
39.     def initialize(self):
    
40.         # The two measures to be joined will be stored togheter in an array,
    
41.         # where the inf value means that the measure is missing
    
42.         default = lambda: array('f', [float('inf')] * 2)
    
43. 
    
44.         # The array will be stored in a dictionary using as key the sensor_id
    
45.         # which produced the measures
    
46.         self.join = defaultdict(default)
    
47. 
    
48.     def process_tick(self):
    
49.         for idx, vals in self.join.iteritems():
    
50.             if vals[0] != float('inf') and vals[1] != float('inf'):
    
51.                 log.debug("id: {0}, heat-index: {1} F"
    
52.                           .format(idx, _heat_index(*vals)))
    
53.         self.join.clear()
    
54. 
    
55.     def process_tuple(self, tup):
    
56.         measure = Measure(*tup.values)
    
57. 
    
58.         # accept only measures within the update time window
    
59.         if measure.timestamp < time.time() - self.conf.tick_tuple_freq:
    
60.             return
    
61. 
    
62.         # Since all components of this topology have a single output stream,
    
63.         # the component id can be used to discriminate between the two streams
    
64.         if tup.comp == "temperature-monitor":
    
65.             if measure.measure > 80:
    
66.                 self.join[measure.id_sensor][0] = measure.measure
    
67.         elif tup.comp == "humidity-monitor":
    
68.             if measure.measure > 40:
    
69.                 self.join[measure.id_sensor][1] = measure.measure
    
70.         else:
    
71.             raise ValueError("Unexpected component stream: {0}"
    
72.                              .format(tup.comp))
    
73. 
    
74. if __name__ == '__main__':
    
75.     logging.basicConfig(
    
76.         level=logging.DEBUG,
    
77.         filename='/tmp/apparent_temperature_heat_index.log',
    
78.         filemode='a',
    
79.     )
    
80. 
    
81.     HeatIndexBolt().run()

复制代码

谢谢楼主分享！