src/main/java/com/zdjizhi/common/ArangodbDnsWindow.java


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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);
    }
}