优化代码：去除无使用的类

22 files changed, 56 insertions, 254 deletions

diff --git a/src/main/java/com/zdjizhi/common/FlowWriteConfig.java b/src/main/java/com/zdjizhi/common/FlowWriteConfig.java
index 57f09a0..3b68285 100644
--- a/src/main/java/com/zdjizhi/common/FlowWriteConfig.java
+++ b/src/main/java/com/zdjizhi/common/FlowWriteConfig.java
@@ -39,17 +39,6 @@ public class FlowWriteConfig {
     public static final String ENCODING = "UTF8";
 
     /**
-     * Nacos
-     */
-    public static final String NACOS_SERVER = FlowWriteConfigurations.getStringProperty(0, "nacos.server");
-    public static final String NACOS_SCHEMA_NAMESPACE = FlowWriteConfigurations.getStringProperty(0, "nacos.schema.namespace");
-    public static final String NACOS_COMMON_NAMESPACE = FlowWriteConfigurations.getStringProperty(0, "nacos.common.namespace");
-    public static final String NACOS_DATA_ID = FlowWriteConfigurations.getStringProperty(0, "nacos.data.id");
-    public static final String NACOS_PIN = FlowWriteConfigurations.getStringProperty(1, "nacos.pin");
-    public static final String NACOS_GROUP = FlowWriteConfigurations.getStringProperty(1, "nacos.group");
-    public static final String NACOS_USERNAME = FlowWriteConfigurations.getStringProperty(1, "nacos.username");
-
-    /**
      * System config
      */
     public static final Integer SOURCE_PARALLELISM = FlowWriteConfigurations.getIntProperty(0, "source.parallelism");
@@ -62,12 +51,6 @@ public class FlowWriteConfig {
     public static final Integer BUFFER_TIMEOUT = FlowWriteConfigurations.getIntProperty(1, "buffer.timeout");
 
     /**
-     * HBase
-     */
-    public static final Integer HBASE_TICK_TUPLE_FREQ_SECS = FlowWriteConfigurations.getIntProperty(0, "hbase.tick.tuple.freq.secs");
-    public static final String HBASE_TABLE_NAME = FlowWriteConfigurations.getStringProperty(1, "hbase.table.name");
-
-    /**
      * kafka common
      */
     public static final String KAFKA_SASL_JAAS_USER = encryptor.decrypt(FlowWriteConfigurations.getStringProperty(1, "kafka.user"));
@@ -76,7 +59,6 @@ public class FlowWriteConfig {
     /**
      * kafka source config
      */
-    public static final String SOURCE_KAFKA_TOPIC = FlowWriteConfigurations.getStringProperty(0, "source.kafka.topic");
     public static final String GROUP_ID = FlowWriteConfigurations.getStringProperty(0, "group.id");
     public static final String SESSION_TIMEOUT_MS = FlowWriteConfigurations.getStringProperty(1, "session.timeout.ms");
     public static final String MAX_POLL_RECORDS = FlowWriteConfigurations.getStringProperty(1, "max.poll.records");
@@ -85,7 +67,6 @@ public class FlowWriteConfig {
     /**
      * kafka sink config
      */
-    public static final String SINK_KAFKA_TOPIC = FlowWriteConfigurations.getStringProperty(0, "sink.kafka.topic");
     public static final String PRODUCER_ACK = FlowWriteConfigurations.getStringProperty(1, "producer.ack");
     public static final String PRODUCER_KAFKA_COMPRESSION_TYPE = FlowWriteConfigurations.getStringProperty(0, "producer.kafka.compression.type");
 
@@ -100,19 +81,12 @@ public class FlowWriteConfig {
     public static final Integer MAX_REQUEST_SIZE = FlowWriteConfigurations.getIntProperty(1, "max.request.size");
 
     /**
-     * http
-     */
-    public static final String APP_ID_HTTP = FlowWriteConfigurations.getStringProperty(1, "app.id.http");
-    public static final Integer APP_TICK_TUPLE_FREQ_SECS = FlowWriteConfigurations.getIntProperty(1, "app.tick.tuple.freq.secs");
-
-    /**
      * common config
      */
     public static final String SOURCE_KAFKA_SERVERS = FlowWriteConfigurations.getStringProperty(0, "source.kafka.servers");
     public static final String SINK_KAFKA_SERVERS = FlowWriteConfigurations.getStringProperty(0, "sink.kafka.servers");
     public static final String ZOOKEEPER_SERVERS = FlowWriteConfigurations.getStringProperty(0, "zookeeper.servers");
     public static final String TOOLS_LIBRARY = FlowWriteConfigurations.getStringProperty(0, "tools.library");
-    public static final String HBASE_ZOOKEEPER_SERVERS = FlowWriteConfigurations.getStringProperty(0, "hbase.zookeeper.servers");
 
 
     /*
@@ -125,7 +99,7 @@ public class FlowWriteConfig {
     public static final int CK_CONNECTION_TIMEOUT = FlowWriteConfigurations.getIntProperty(0, "ck.connection.timeout");
     public static final int CK_SOCKET_TIMEOUT = FlowWriteConfigurations.getIntProperty(0, "ck.socket.timeout");
 
-    public static final int FLINK_WATERMARK_MAX_ORDERNESS = FlowWriteConfigurations.getIntProperty(0, "flink.watermark.max.orderness");
+    public static final int FLINK_WATERMARK_MAX_DELAY_TIME = FlowWriteConfigurations.getIntProperty(0, "flink.watermark.max.delay.time");
     public static final int LOG_AGGREGATE_DURATION = FlowWriteConfigurations.getIntProperty(0, "log.aggregate.duration");
     public static final int LOG_AGGREGATE_DURATION_GRAPH = FlowWriteConfigurations.getIntProperty(0, "log.aggregate.duration.graph");
     public static final String SOURCE_KAFKA_TOPIC_DNS = FlowWriteConfigurations.getStringProperty(0, "source.kafka.topic.dns");
@@ -147,8 +121,6 @@ public class FlowWriteConfig {
     public static final String R_NX_DOMAIN2DOMAIN = FlowWriteConfigurations.getStringProperty(0, "sink.arango.table.r.nx.domain2domain");
 
 
-
-
     public static final String ARANGODB_HOST = FlowWriteConfigurations.getStringProperty(0, "arangoDB.host");
     public static final Integer ARANGODB_PORT = FlowWriteConfigurations.getIntProperty(0, "arangoDB.port");
     public static final String ARANGODB_USER = FlowWriteConfigurations.getStringProperty(0, "arangoDB.user");
@@ -156,11 +128,8 @@ public class FlowWriteConfig {
     public static final String ARANGODB_DB_NAME = FlowWriteConfigurations.getStringProperty(0, "arangoDB.DB.name");
     public static final Integer ARANGODB_TTL = FlowWriteConfigurations.getIntProperty(0, "arangoDB.ttl");
     public static final Integer ARANGODB_BATCH = FlowWriteConfigurations.getIntProperty(0, "arangoDB.batch");
-
-    public static final Integer UPDATE_ARANGO_BATCH = FlowWriteConfigurations.getIntProperty(0, "update.arango.batch");
-    public static final String ARANGODB_READ_LIMIT = FlowWriteConfigurations.getStringProperty(0, "arangoDB.read.limit");
     public static final Integer THREAD_POOL_NUMBER = FlowWriteConfigurations.getIntProperty(0, "thread.pool.number");
-    public static final Integer THREAD_AWAIT_TERMINATION_TIME = FlowWriteConfigurations.getIntProperty(0, "thread.await.termination.time");
-    public static final Integer SINK_BATCH_TIME_OUT = FlowWriteConfigurations.getIntProperty(0, "sink.batch.time.out");
-    public static final Integer SINK_BATCH = FlowWriteConfigurations.getIntProperty(0, "sink.batch");
+
+    public static final Integer SINK_BATCH_DELAY_TIME = FlowWriteConfigurations.getIntProperty(0, "sink.batch.delay.time");
+    public static final Integer CK_BATCH = FlowWriteConfigurations.getIntProperty(0, "ck.batch");
 }
\ No newline at end of file
diff --git a/src/main/java/com/zdjizhi/common/CKWindow.java b/src/main/java/com/zdjizhi/etl/CKBatchWindow.java
index b7c7b8c..f66455f 100644
--- a/src/main/java/com/zdjizhi/common/CKWindow.java
+++ b/src/main/java/com/zdjizhi/etl/CKBatchWindow.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.common;
+package com.zdjizhi.etl;
 
 import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
@@ -9,7 +9,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
-public class CKWindow implements AllWindowFunction<Map<String, Object>, List<Map<String, Object>>, TimeWindow> {
+public class CKBatchWindow implements AllWindowFunction<Map<String, Object>, List<Map<String, Object>>, TimeWindow> {
 
     @Override
     public void apply(TimeWindow timeWindow, Iterable<Map<String, Object>> iterable, Collector<List<Map<String, Object>>> out) throws Exception {
diff --git a/src/main/java/com/zdjizhi/common/ArangodbIPWindow.java b/src/main/java/com/zdjizhi/etl/connection/ArangodbBatchIPWindow.java
index f91fb08..f5848c4 100644
--- a/src/main/java/com/zdjizhi/common/ArangodbIPWindow.java
+++ b/src/main/java/com/zdjizhi/etl/connection/ArangodbBatchIPWindow.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.common;
+package com.zdjizhi.etl.connection;
 
 import cn.hutool.core.util.StrUtil;
 import com.arangodb.entity.BaseEdgeDocument;
@@ -11,7 +11,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
-public class ArangodbIPWindow implements AllWindowFunction<Map<String, Object>, List<BaseEdgeDocument>, TimeWindow> {
+public class ArangodbBatchIPWindow implements AllWindowFunction<Map<String, Object>, List<BaseEdgeDocument>, TimeWindow> {
 
     @Override
     public void apply(TimeWindow timeWindow, Iterable<Map<String, Object>> iterable, Collector<List<BaseEdgeDocument>> collector) throws Exception {
diff --git a/src/main/java/com/zdjizhi/etl/ConnProcessFunction.java b/src/main/java/com/zdjizhi/etl/connection/ConnProcessFunction.java
index 49041dc..cec2425 100644
--- a/src/main/java/com/zdjizhi/etl/ConnProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/connection/ConnProcessFunction.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.etl;
+package com.zdjizhi.etl.connection;
 
 import cn.hutool.core.convert.Convert;
 import cn.hutool.log.Log;
@@ -24,21 +24,8 @@ public class ConnProcessFunction extends ProcessWindowFunction<Map<String, Objec
 
     @Override
     public void process(Tuple2<String, String> keys, Context context, Iterable<Map<String, Object>> elements, Collector<Map<String, Object>> out) {
-        Map<String, Object> middleResult = getMiddleResult(keys, elements);
-        try {
-            if (middleResult != null) {
-                out.collect(middleResult);
-                logger.debug("获取中间聚合结果：{}", middleResult.toString());
-            }
-        } catch (Exception e) {
-            logger.error("获取中间聚合结果失败,middleResult: {}\n{}", middleResult.toString(), e);
-        }
-    }
-
-    private Map<String, Object> getMiddleResult(Tuple2<String, String> keys, Iterable<Map<String, Object>> elements) {
-
-        Tuple5<Long, Long, Long, Long, Long> values = connAggregate(elements);
         try {
+            Tuple5<Long, Long, Long, Long, Long> values = connAggregate(elements);
             if (values != null) {
                 Map<String, Object> result = new LinkedHashMap<>();
                 result.put("start_time", values.f0);
@@ -48,13 +35,12 @@ public class ConnProcessFunction extends ProcessWindowFunction<Map<String, Objec
                 result.put("sessions", values.f2);
                 result.put("packets", values.f3);
                 result.put("bytes", values.f4);
-                return result;
+                out.collect(result);
+                logger.debug("获取中间聚合结果：{}", result.toString());
             }
-
         } catch (Exception e) {
-            logger.error("加载中间结果集失败,keys: {} values: {}\n{}", keys, values, e);
+            logger.error("获取中间聚合结果失败,middleResult: {}", e);
         }
-        return null;
     }
 
     private Tuple5<Long, Long, Long, Long, Long> connAggregate(Iterable<Map<String, Object>> elements) {
diff --git a/src/main/java/com/zdjizhi/etl/Ip2IpGraphProcessFunction.java b/src/main/java/com/zdjizhi/etl/connection/Ip2IpGraphProcessFunction.java
index 5bce25d..e9dd0e2 100644
--- a/src/main/java/com/zdjizhi/etl/Ip2IpGraphProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/connection/Ip2IpGraphProcessFunction.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.etl;
+package com.zdjizhi.etl.connection;
 
 import cn.hutool.core.convert.Convert;
 import cn.hutool.core.date.DateUtil;
diff --git a/src/main/java/com/zdjizhi/common/IpKeysSelector.java b/src/main/java/com/zdjizhi/etl/connection/IpKeysSelector.java
index 2dc7e79..470648e 100644
--- a/src/main/java/com/zdjizhi/common/IpKeysSelector.java
+++ b/src/main/java/com/zdjizhi/etl/connection/IpKeysSelector.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.common;
+package com.zdjizhi.etl.connection;
 
 
 import org.apache.flink.api.java.functions.KeySelector;
diff --git a/src/main/java/com/zdjizhi/etl/SketchProcessFunction.java b/src/main/java/com/zdjizhi/etl/connection/SketchProcessFunction.java
index 54d53b6..698ed55 100644
--- a/src/main/java/com/zdjizhi/etl/SketchProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/connection/SketchProcessFunction.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.etl;
+package com.zdjizhi.etl.connection;
 
 import cn.hutool.core.convert.Convert;
 import cn.hutool.log.Log;
diff --git a/src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java b/src/main/java/com/zdjizhi/etl/dns/ArangodbBatchDnsWindow.java
index 373eef5..ada4b83 100644
--- a/src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java
+++ b/src/main/java/com/zdjizhi/etl/dns/ArangodbBatchDnsWindow.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.common;
+package com.zdjizhi.etl.dns;
 
 import cn.hutool.core.util.StrUtil;
 import com.arangodb.entity.BaseEdgeDocument;
@@ -11,7 +11,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
-public class ArangodbDnsWindow implements AllWindowFunction<Map<String, Object>, List<BaseEdgeDocument>, TimeWindow> {
+public class ArangodbBatchDnsWindow implements AllWindowFunction<Map<String, Object>, List<BaseEdgeDocument>, TimeWindow> {
 
     @Override
     public void apply(TimeWindow timeWindow, Iterable<Map<String, Object>> iterable, Collector<List<BaseEdgeDocument>> out) throws Exception {
diff --git a/src/main/java/com/zdjizhi/common/DnsGraphKeysSelector.java b/src/main/java/com/zdjizhi/etl/dns/DnsGraphKeysSelector.java
index 5aa08c5..f6e6a6f 100644
--- a/src/main/java/com/zdjizhi/common/DnsGraphKeysSelector.java
+++ b/src/main/java/com/zdjizhi/etl/dns/DnsGraphKeysSelector.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.common;
+package com.zdjizhi.etl.dns;
 
 import cn.hutool.core.util.StrUtil;
 import org.apache.flink.api.java.functions.KeySelector;
diff --git a/src/main/java/com/zdjizhi/etl/DnsGraphProcessFunction.java b/src/main/java/com/zdjizhi/etl/dns/DnsGraphProcessFunction.java
index 18d7a71..d8b3a36 100644
--- a/src/main/java/com/zdjizhi/etl/DnsGraphProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/dns/DnsGraphProcessFunction.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.etl;
+package com.zdjizhi.etl.dns;
 
 import cn.hutool.core.convert.Convert;
 import cn.hutool.log.Log;
diff --git a/src/main/java/com/zdjizhi/etl/DnsMapFunction.java b/src/main/java/com/zdjizhi/etl/dns/DnsMapFunction.java
index 86c4616..b536152 100644
--- a/src/main/java/com/zdjizhi/etl/DnsMapFunction.java
+++ b/src/main/java/com/zdjizhi/etl/dns/DnsMapFunction.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.etl;
+package com.zdjizhi.etl.dns;
 
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONArray;
@@ -47,7 +47,7 @@ public class DnsMapFunction implements MapFunction<Map<String, Object>, Map<Stri
                 } else if (DnsType.CNAME.getCode().equals(type)) {
                     dnsCNAME = Joiner.on(",").skipNulls().join(dnsCNAME, body);
                     dnsCNAMENum++;
-                } else if (DnsType.CNAME.getCode().equals(type)) {
+                } else if (DnsType.NS.getCode().equals(type)) {
                     dnsNs = Joiner.on(",").skipNulls().join(dnsNs, body);
                     dnsNsNum++;
                 } else if (DnsType.MX.getCode().equals(type)) {
diff --git a/src/main/java/com/zdjizhi/etl/DnsRelationProcessFunction.java b/src/main/java/com/zdjizhi/etl/dns/DnsRelationProcessFunction.java
index 04e45f8..03a0c41 100644
--- a/src/main/java/com/zdjizhi/etl/DnsRelationProcessFunction.java
+++ b/src/main/java/com/zdjizhi/etl/dns/DnsRelationProcessFunction.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.etl;
+package com.zdjizhi.etl.dns;
 
 import cn.hutool.core.convert.Convert;
 import cn.hutool.core.date.DateUtil;
diff --git a/src/main/java/com/zdjizhi/etl/DnsSplitFlatMapFunction.java b/src/main/java/com/zdjizhi/etl/dns/DnsSplitFlatMapFunction.java
index 6dbe35e..96ce7b9 100644
--- a/src/main/java/com/zdjizhi/etl/DnsSplitFlatMapFunction.java
+++ b/src/main/java/com/zdjizhi/etl/dns/DnsSplitFlatMapFunction.java
@@ -1,4 +1,4 @@
-package com.zdjizhi.etl;
+package com.zdjizhi.etl.dns;
 
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
diff --git a/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java b/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java
index 0ed5052..7ca7acb 100644
--- a/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java
+++ b/src/main/java/com/zdjizhi/topology/LogFlowWriteTopology.java
@@ -4,9 +4,11 @@ import cn.hutool.core.convert.Convert;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
-import com.zdjizhi.common.*;
+import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.enums.DnsType;
-import com.zdjizhi.etl.*;
+import com.zdjizhi.etl.CKBatchWindow;
+import com.zdjizhi.etl.connection.*;
+import com.zdjizhi.etl.dns.*;
 import com.zdjizhi.utils.arangodb.ArangoDBSink;
 import com.zdjizhi.utils.ck.ClickhouseSink;
 import com.zdjizhi.utils.kafka.KafkaConsumer;
@@ -49,7 +51,7 @@ public class LogFlowWriteTopology {
                 //transform
                 DataStream<Map<String, Object>> connTransformStream = connSource
                         .assignTimestampsAndWatermarks(WatermarkStrategy
-                                .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(FlowWriteConfig.FLINK_WATERMARK_MAX_ORDERNESS))
+                                .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(FlowWriteConfig.FLINK_WATERMARK_MAX_DELAY_TIME))
                                 .withTimestampAssigner((event, timestamp) -> Convert.toLong(event.get("conn_start_time")) * 1000))
                         .keyBy(new IpKeysSelector())
                         .window(TumblingProcessingTimeWindows.of(Time.seconds(FlowWriteConfig.LOG_AGGREGATE_DURATION)))
@@ -58,7 +60,7 @@ public class LogFlowWriteTopology {
                         .setParallelism(FlowWriteConfig.TRANSFORM_PARALLELISM);
 
                 DataStream<Map<String, Object>> sketchTransformStream = sketchSource.assignTimestampsAndWatermarks(WatermarkStrategy
-                        .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(FlowWriteConfig.FLINK_WATERMARK_MAX_ORDERNESS))
+                        .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(FlowWriteConfig.FLINK_WATERMARK_MAX_DELAY_TIME))
                         .withTimestampAssigner((event, timestamp) -> Convert.toLong(event.get("sketch_start_time")) * 1000))
                         .keyBy(new IpKeysSelector())
                         .window(TumblingProcessingTimeWindows.of(Time.seconds(FlowWriteConfig.LOG_AGGREGATE_DURATION)))
@@ -72,16 +74,16 @@ public class LogFlowWriteTopology {
                         .keyBy(new IpKeysSelector())
                         .window(TumblingProcessingTimeWindows.of(Time.seconds(LOG_AGGREGATE_DURATION_GRAPH)))
                         .process(new Ip2IpGraphProcessFunction())
-//                        .filter(Objects::nonNull)
+                        .filter(Objects::nonNull)
                         .setParallelism(TRANSFORM_PARALLELISM);
 
                 //写入CKsink,批量处理
-                connSource.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_TIME_OUT))).apply(new CKWindow()).addSink(new ClickhouseSink(SINK_CK_TABLE_CONNECTION)).name("CKSink");
-                sketchSource.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_TIME_OUT))).apply(new CKWindow()).addSink(new ClickhouseSink(SINK_CK_TABLE_SKETCH)).name("CKSink");
-                sketchTransformStream.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_TIME_OUT))).apply(new CKWindow()).addSink(new ClickhouseSink(SINK_CK_TABLE_RELATION_CONNECTION)).name("CKSink");
+                connSource.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_DELAY_TIME))).apply(new CKBatchWindow()).addSink(new ClickhouseSink(SINK_CK_TABLE_CONNECTION)).name("CKSink");
+                sketchSource.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_DELAY_TIME))).apply(new CKBatchWindow()).addSink(new ClickhouseSink(SINK_CK_TABLE_SKETCH)).name("CKSink");
+                sketchTransformStream.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_DELAY_TIME))).apply(new CKBatchWindow()).addSink(new ClickhouseSink(SINK_CK_TABLE_RELATION_CONNECTION)).name("CKSink");
 
                 //写入arangodb
-                ip2ipGraph.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_TIME_OUT))).apply(new ArangodbIPWindow()).addSink(new ArangoDBSink(R_VISIT_IP2IP)).name(R_VISIT_IP2IP);
+                ip2ipGraph.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_DELAY_TIME))).apply(new ArangodbBatchIPWindow()).addSink(new ArangoDBSink(R_VISIT_IP2IP)).name(R_VISIT_IP2IP);
 
             } else if (FlowWriteConfig.LOG_NEED_COMPLETE == 2) {
 
@@ -92,7 +94,7 @@ public class LogFlowWriteTopology {
                         .name(FlowWriteConfig.SOURCE_KAFKA_TOPIC_DNS);
 
                 DataStream<Map<String, Object>> dnsTransform = dnsSource.assignTimestampsAndWatermarks(WatermarkStrategy
-                        .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(FLINK_WATERMARK_MAX_ORDERNESS))
+                        .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(FLINK_WATERMARK_MAX_DELAY_TIME))
                         .withTimestampAssigner((event, timestamp) -> Convert.toLong(event.get("capture_time")) * 1000))
                         .flatMap(new DnsSplitFlatMapFunction())
                         .keyBy(new DnsGraphKeysSelector())
@@ -104,13 +106,13 @@ public class LogFlowWriteTopology {
                 //dns 原始日志 ck入库
                 dnsSource.filter(Objects::nonNull)
                         .setParallelism(FlowWriteConfig.SINK_PARALLELISM)
-                        .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_TIME_OUT))).apply(new CKWindow())
+                        .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_DELAY_TIME))).apply(new CKBatchWindow())
                         .addSink(new ClickhouseSink(SINK_CK_TABLE_DNS))
                         .setParallelism(FlowWriteConfig.SINK_PARALLELISM)
                         .name("CKSink");
 
                 //dns 拆分后relation日志 ck入库
-                dnsTransform.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_TIME_OUT))).apply(new CKWindow())
+                dnsTransform.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_DELAY_TIME))).apply(new CKBatchWindow())
                         .addSink(new ClickhouseSink(SINK_CK_TABLE_RELATION_DNS))
                         .setParallelism(SINK_PARALLELISM)
                         .name("CKSink");
@@ -119,12 +121,11 @@ public class LogFlowWriteTopology {
                 DataStream<Map<String, Object>> dnsGraph = dnsTransform.keyBy(new DnsGraphKeysSelector())
                         .window(TumblingProcessingTimeWindows.of(Time.seconds(LOG_AGGREGATE_DURATION_GRAPH)))
                         .process(new DnsGraphProcessFunction())
-                        .setParallelism(SINK_PARALLELISM)
-                        .filter(Objects::nonNull);
+                        .setParallelism(SINK_PARALLELISM);
 
                 for (DnsType dnsEnum : DnsType.values()) {
                     dnsGraph.filter(x -> ObjectUtil.equal(dnsEnum.getType(), x.get("record_type")))
-                            .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_TIME_OUT))).apply(new ArangodbDnsWindow())
+                            .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(SINK_BATCH_DELAY_TIME))).apply(new ArangodbBatchDnsWindow())
                             .addSink(new ArangoDBSink(dnsEnum.getSink()))
                             .setParallelism(SINK_PARALLELISM)
                             .name("ArangodbSink");
diff --git a/src/main/java/com/zdjizhi/utils/app/AppUtils.java b/src/main/java/com/zdjizhi/utils/app/AppUtils.java
deleted file mode 100644
index 1425ce9..0000000
--- a/src/main/java/com/zdjizhi/utils/app/AppUtils.java
+++ /dev/null
@@ -1,124 +0,0 @@
-package com.zdjizhi.utils.app;
-
-import cn.hutool.log.Log;
-import cn.hutool.log.LogFactory;
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
-import com.zdjizhi.common.FlowWriteConfig;
-import com.zdjizhi.utils.StringUtil;
-import com.zdjizhi.utils.http.HttpClientUtil;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-/**
- * AppId 工具类
- *
- * @author qidaijie
- */
-
-@Deprecated
-public class AppUtils {
-    private static final Log logger = LogFactory.get();
-    private static Map<Integer, String> appIdMap = new ConcurrentHashMap<>(128);
-    private static AppUtils appUtils;
-
-    private static void getAppInstance() {
-        appUtils = new AppUtils();
-    }
-
-
-    /**
-     * 构造函数-新
-     */
-    private AppUtils() {
-        //定时更新
-        updateAppIdCache();
-    }
-
-    /**
-     * 更新变量
-     */
-    private static void change() {
-        if (appUtils == null) {
-            getAppInstance();
-        }
-        timestampsFilter();
-    }
-
-
-    /**
-     * 获取变更内容
-     */
-    private static void timestampsFilter() {
-        try {
-            Long begin = System.currentTimeMillis();
-            String schema = HttpClientUtil.requestByGetMethod(FlowWriteConfig.APP_ID_HTTP);
-            if (StringUtil.isNotBlank(schema)) {
-                String data = JSONObject.parseObject(schema).getString("data");
-                JSONArray objects = JSONArray.parseArray(data);
-                for (Object object : objects) {
-                    JSONArray jsonArray = JSONArray.parseArray(object.toString());
-                    int key = jsonArray.getInteger(0);
-                    String value = jsonArray.getString(1);
-                    if (appIdMap.containsKey(key)) {
-                        if (!value.equals(appIdMap.get(key))) {
-                            appIdMap.put(key, value);
-                        }
-                    } else {
-                        appIdMap.put(key, value);
-                    }
-                }
-                logger.warn("Updating the correspondence takes time:" + (begin - System.currentTimeMillis()));
-                logger.warn("Pull the length of the interface data:[" + objects.size() + "]");
-            }
-        } catch (RuntimeException e) {
-            logger.error("Update cache app-id failed, exception：" + e);
-        }
-    }
-
-
-    /**
-     * 验证定时器,每隔一段时间验证一次-验证获取新的Cookie
-     */
-    private void updateAppIdCache() {
-        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);
-        executorService.scheduleAtFixedRate(new Runnable() {
-            @Override
-            public void run() {
-                try {
-                    if (FlowWriteConfig.APP_TICK_TUPLE_FREQ_SECS != 0) {
-                        change();
-                    }
-                } catch (RuntimeException e) {
-                    logger.error("AppUtils update AppCache is error===>{" + e + "}<===");
-                }
-            }
-        }, 1, FlowWriteConfig.APP_TICK_TUPLE_FREQ_SECS, TimeUnit.SECONDS);
-    }
-
-
-    /**
-     * 获取 appName
-     *
-     * @param appId app_id
-     * @return account
-     */
-    public static String getAppName(int appId) {
-
-        if (appUtils == null) {
-            getAppInstance();
-        }
-
-        if (appIdMap.containsKey(appId)) {
-            return appIdMap.get(appId);
-        } else {
-            logger.warn("AppMap get appName is null, ID is :" + appId);
-            return "";
-        }
-    }
-
-}
diff --git a/src/main/java/com/zdjizhi/utils/arangodb/ArangoDBSink.java b/src/main/java/com/zdjizhi/utils/arangodb/ArangoDBSink.java
index d2306b7..d94dad7 100644
--- a/src/main/java/com/zdjizhi/utils/arangodb/ArangoDBSink.java
+++ b/src/main/java/com/zdjizhi/utils/arangodb/ArangoDBSink.java
@@ -17,9 +17,8 @@ public class ArangoDBSink extends RichSinkFunction<List<BaseEdgeDocument>> {
     private String collection;
 
     @Override
-    public void invoke(List<BaseEdgeDocument> BaseEdgeDocuments, Context context) throws Exception {
-
-        arangoDBConnect.overwrite(BaseEdgeDocuments, getCollection());
+    public void invoke(List<BaseEdgeDocument> baseEdgeDocuments, Context context) throws Exception {
+        arangoDBConnect.overwrite(baseEdgeDocuments, getCollection());
     }
 
     @Override
@@ -47,3 +46,4 @@ public class ArangoDBSink extends RichSinkFunction<List<BaseEdgeDocument>> {
     }
 
 }
+
diff --git a/src/main/java/com/zdjizhi/utils/ck/ClickhouseSink.java b/src/main/java/com/zdjizhi/utils/ck/ClickhouseSink.java
index cb37500..d3e14cc 100644
--- a/src/main/java/com/zdjizhi/utils/ck/ClickhouseSink.java
+++ b/src/main/java/com/zdjizhi/utils/ck/ClickhouseSink.java
@@ -29,7 +29,6 @@ public class ClickhouseSink extends RichSinkFunction<List<Map<String, Object>>>
     private PreparedStatement preparedStatement;
     public String sink;
 
-
     public ClickhouseSink(String sink) {
         this.sink = sink;
     }
@@ -49,7 +48,7 @@ public class ClickhouseSink extends RichSinkFunction<List<Map<String, Object>>>
 
     @Override
     public void open(Configuration parameters) throws Exception {
-
+        super.open(parameters);
         try {
 //            Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
 //            connection = DriverManager.getConnection("jdbc:clickhouse://" + CK_HOSTS + "/" + CK_DATABASE, CK_USERNAME, CK_PIN);
@@ -64,8 +63,7 @@ public class ClickhouseSink extends RichSinkFunction<List<Map<String, Object>>>
             BalancedClickhouseDataSource dataSource = new BalancedClickhouseDataSource("jdbc:clickhouse://" + CK_HOSTS, properties);
             dataSource.scheduleActualization(10, TimeUnit.SECONDS);//开启检测
             connection = dataSource.getConnection();
-
-            log.info("get clickhouse connection success");
+            log.debug("get clickhouse connection success");
         } catch (SQLException e) {
             log.error("clickhouse connection error ,{}", e);
         }
@@ -102,7 +100,7 @@ public class ClickhouseSink extends RichSinkFunction<List<Map<String, Object>>>
                 preparedStatement.addBatch();
                 count++;
                 //1w提交一次
-                if (count % SINK_BATCH == 0) {
+                if (count % CK_BATCH == 0) {
                     preparedStatement.executeBatch();
                     connection.commit();
                     preparedStatement.clearBatch();
diff --git a/src/main/java/com/zdjizhi/utils/general/TransFormMap.java b/src/main/java/com/zdjizhi/utils/general/TransFormMap.java
index 27daa71..42b7639 100644
--- a/src/main/java/com/zdjizhi/utils/general/TransFormMap.java
+++ b/src/main/java/com/zdjizhi/utils/general/TransFormMap.java
@@ -113,11 +113,6 @@ public class TransFormMap {
                     JsonParseUtil.setValue(jsonMap, appendToKeyName, TransFunction.flattenSpec(logValue.toString(), param));
                 }
                 break;
-            case "app_match":
-                if (logValue != null && appendTo == null) {
-                    JsonParseUtil.setValue(jsonMap, appendToKeyName, TransFunction.appMatch(logValue.toString()));
-                }
-                break;
             default:
         }
     }
diff --git a/src/main/java/com/zdjizhi/utils/general/TransFormTypeMap.java b/src/main/java/com/zdjizhi/utils/general/TransFormTypeMap.java
index 34cabfa..9e92576 100644
--- a/src/main/java/com/zdjizhi/utils/general/TransFormTypeMap.java
+++ b/src/main/java/com/zdjizhi/utils/general/TransFormTypeMap.java
@@ -115,11 +115,6 @@ public class TransFormTypeMap {
                     JsonParseUtil.setValue(jsonMap, appendToKeyName, TransFunction.flattenSpec(logValue.toString(), param));
                 }
                 break;
-            case "app_match":
-                if (logValue != null && appendToKeyValue == null) {
-                    JsonParseUtil.setValue(jsonMap, appendToKeyName, TransFunction.appMatch(logValue.toString()));
-                }
-                break;
             default:
         }
     }
diff --git a/src/main/java/com/zdjizhi/utils/general/TransFunction.java b/src/main/java/com/zdjizhi/utils/general/TransFunction.java
index e3363f9..84fe5cc 100644
--- a/src/main/java/com/zdjizhi/utils/general/TransFunction.java
+++ b/src/main/java/com/zdjizhi/utils/general/TransFunction.java
@@ -1,7 +1,6 @@
 package com.zdjizhi.utils.general;
 
 import cn.hutool.core.codec.Base64;
-import cn.hutool.core.util.StrUtil;
 import cn.hutool.log.Log;
 import cn.hutool.log.LogFactory;
 import com.jayway.jsonpath.InvalidPathException;
@@ -10,7 +9,6 @@ import com.zdjizhi.common.FlowWriteConfig;
 import com.zdjizhi.utils.FormatUtils;
 import com.zdjizhi.utils.IpLookupV2;
 import com.zdjizhi.utils.StringUtil;
-import com.zdjizhi.utils.app.AppUtils;
 import com.zdjizhi.utils.json.JsonParseUtil;
 
 import java.math.BigInteger;
@@ -126,22 +124,6 @@ class TransFunction {
      * @return account
      */
 
-    /**
-     * appId与缓存中对应关系补全appName
-     *
-     * @param appIds app id 列表
-     * @return appName
-     */
-    @Deprecated
-    static String appMatch(String appIds) {
-        try {
-            String appId = StrUtil.split(appIds, FlowWriteConfig.FORMAT_SPLITTER, true, true).get(0);
-            return AppUtils.getAppName(Integer.parseInt(appId));
-        } catch (NumberFormatException | ClassCastException exception) {
-            logger.error("APP ID列表分割转换异常，异常APP ID列表:" + appIds);
-            return "";
-        }
-    }
 
     /**
      * 解析顶级域名
diff --git a/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java b/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java
index 97a53da..e15d535 100644
--- a/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java
+++ b/src/main/java/com/zdjizhi/utils/kafka/KafkaConsumer.java
@@ -14,10 +14,10 @@ import java.util.Properties;
  * @date 2021/6/813:54
  */
 public class KafkaConsumer {
-    private static Properties createConsumerConfig() {
+    private static Properties createConsumerConfig(String topic) {
         Properties properties = new Properties();
         properties.put("bootstrap.servers", FlowWriteConfig.SOURCE_KAFKA_SERVERS);
-        properties.put("group.id", FlowWriteConfig.GROUP_ID);
+        properties.put("group.id", FlowWriteConfig.GROUP_ID + "-" + topic);
         properties.put("session.timeout.ms", FlowWriteConfig.SESSION_TIMEOUT_MS);
         properties.put("max.poll.records", FlowWriteConfig.MAX_POLL_RECORDS);
         properties.put("max.partition.fetch.bytes", FlowWriteConfig.MAX_PARTITION_FETCH_BYTES);
@@ -33,9 +33,9 @@ public class KafkaConsumer {
      * @return kafka logs -> map
      */
     @SuppressWarnings("unchecked")
-    public static FlinkKafkaConsumer<Map<String,Object>> myDeserializationConsumer(String topic) {
-        FlinkKafkaConsumer<Map<String,Object>> kafkaConsumer = new FlinkKafkaConsumer<>(topic,
-                new TimestampDeserializationSchema(), createConsumerConfig());
+    public static FlinkKafkaConsumer<Map<String, Object>> myDeserializationConsumer(String topic) {
+        FlinkKafkaConsumer<Map<String, Object>> kafkaConsumer = new FlinkKafkaConsumer<>(topic,
+                new TimestampDeserializationSchema(), createConsumerConfig(topic));
 
         //随着checkpoint提交，将offset提交到kafka
         kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
@@ -51,9 +51,9 @@ public class KafkaConsumer {
      *
      * @return kafka logs
      */
-    public static FlinkKafkaConsumer<String> flinkConsumer() {
-        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(FlowWriteConfig.SOURCE_KAFKA_TOPIC,
-                new SimpleStringSchema(), createConsumerConfig());
+    public static FlinkKafkaConsumer<String> flinkConsumer(String topic) {
+        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topic,
+                new SimpleStringSchema(), createConsumerConfig(topic));
 
         //随着checkpoint提交，将offset提交到kafka
         kafkaConsumer.setCommitOffsetsOnCheckpoints(true);
diff --git a/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java b/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
index 28ecff9..cf27cc3 100644
--- a/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
+++ b/src/main/java/com/zdjizhi/utils/kafka/KafkaProducer.java
@@ -33,9 +33,9 @@ public class KafkaProducer {
     }
 
 
-    public static FlinkKafkaProducer<String> getKafkaProducer() {
+    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {
         FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<String>(
-                FlowWriteConfig.SINK_KAFKA_TOPIC,
+                topic,
                 new SimpleStringSchema(),
                 createProducerConfig(),
                 //sink与所有分区建立连接，轮询写入；