From 06042db9b11bf3a17eaec455b3daf5b31de679d7 Mon Sep 17 00:00:00 2001
From: zhanghongqing <zhanghongqing@zdjizhi.com>
Date: Tue, 12 Jul 2022 19:24:53 +0800
Subject: 优化代码：使用windowAll做数据批量操作
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../java/com/zdjizhi/common/ArangodbDnsWindow.java | 36 ++++++++++++++++++++++
 1 file changed, 36 insertions(+)
 create mode 100644 src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java

(limited to 'src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java')

diff --git a/src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java b/src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java
new file mode 100644
index 0000000..373eef5
--- /dev/null
+++ b/src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java
@@ -0,0 +1,36 @@
+package com.zdjizhi.common;
+
+import cn.hutool.core.util.StrUtil;
+import com.arangodb.entity.BaseEdgeDocument;
+import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.util.Collector;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+public class ArangodbDnsWindow 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 {
+        Iterator<Map<String, Object>> iterator = iterable.iterator();
+        List<BaseEdgeDocument> batchLog = new ArrayList<>();
+        while (iterator.hasNext()) {
+            Map<String, Object> next = iterator.next();
+            String qname = StrUtil.toString(next.get("qname"));
+            String record = StrUtil.toString(next.get("record"));
+            BaseEdgeDocument baseEdgeDocument = new BaseEdgeDocument();
+            baseEdgeDocument.setKey(String.join("-", qname, record));
+            baseEdgeDocument.setFrom("qname/" + qname);
+            baseEdgeDocument.setTo("record/" + record);
+            baseEdgeDocument.addAttribute("qname", qname);
+            baseEdgeDocument.addAttribute("record", record);
+            baseEdgeDocument.addAttribute("last_found_time", next.get("last_found_time"));
+
+            batchLog.add(baseEdgeDocument);
+        }
+        out.collect(batchLog);
+    }
+}
-- 
cgit v1.2.3