feature: add operator name in LocationMetric

1 files changed, 4 insertions, 2 deletions

diff --git a/module-CN-pre-metrics/src/main/java/com/zdjizhi/pre/location/LocationMetric.java b/module-CN-pre-metrics/src/main/java/com/zdjizhi/pre/location/LocationMetric.java
index 574c256..e552453 100644
--- a/module-CN-pre-metrics/src/main/java/com/zdjizhi/pre/location/LocationMetric.java
+++ b/module-CN-pre-metrics/src/main/java/com/zdjizhi/pre/location/LocationMetric.java
@@ -37,12 +37,14 @@ public class LocationMetric implements Schedule {
 
         SingleOutputStreamOperator<String> locationSubscriberMetric = process.keyBy(LocationSubscriber::getSubscriber_id)
                 .window(TumblingEventTimeWindows.of(Time.minutes(windowsTime)))
-                .reduce(new SecondAggregationReduce(), new MetricSubscriberProcessWindowFunc());
+                .reduce(new SecondAggregationReduce(), new MetricSubscriberProcessWindowFunc())
+                .name("locationSubscriberMetric");
 
         locationSubscriberMetric
                 .addSink(ClickHouseTableFactory.getSinkFunction(Configs.get(CommonConfig.LOCATION_SUBSCRIBER_TABLE)))
                 //.addSink(KafkaUtils.getKafkaSink(Configs.get(CommonConfig.LOCATION_SUBSCRIBER_TOPIC)))
-                .setParallelism(outputParallelism);
+                .setParallelism(outputParallelism)
+                .name("locationSubscriberMetricSink");
 
         //todo GMLC数据源字段详情不清楚，后续再开发GMLC->location_subscriber的数据流
     }