1+ import time
2+ from sage .core .api .service .base_service import BaseService
3+ from sage .core .api .local_environment import LocalEnvironment
4+ from sage .core .api .function .sink_function import SinkFunction
5+ from sage .core .api .function .batch_function import BatchFunction
6+ from sage .common .utils .logging .custom_logger import CustomLogger
7+
8+ # 批处理数据源：生成10条 Hello, World! 数据
9+ class HelloBatch (BatchFunction ):
10+ def __init__ (self , ** kwargs ):
11+ super ().__init__ (** kwargs )
12+ self .counter = 0
13+ self .max_count = 10 # 生成10个数据包后返回None
14+
15+ def execute (self ):
16+ if self .counter >= self .max_count :
17+ return None # 返回None表示批处理完成
18+ self .counter += 1
19+ return f"Hello, World! #{ self .counter }
20+
21+
22+ # 简单 SinkFunction，调用hello服务并打印结果
23+ class PrintSink (SinkFunction ):
24+ def execute (self , data ):
25+ self .call_service ["hello_service" ].hello ()
26+ print (data )
27+
28+
29+ class HelloService (BaseService ):
30+ def __init__ (self ):
31+ self .message = "hello service!!!"
32+
33+ def hello (self ):
34+ print (self .message )
35+
36+
37+ def main ():
38+ env = LocalEnvironment ("hello_service" )
39+ env .register_service ("hello_service" , HelloService )
40+
41+ env .from_batch (HelloBatch ).sink (PrintSink )
42+
43+ try :
44+ print ("Waiting for batch processing to complete..." )
45+ env .submit ()
46+
47+ time .sleep (1 )
48+ except KeyboardInterrupt :
49+ print ("停止运行" )
50+ finally :
51+ print ("Hello Service 批处理示例结束" )
52+
53+ if __name__ == "__main__" :
54+ CustomLogger .disable_global_console_debug ()
55+ main ()
0 commit comments