src/main/java/com/zdjizhi/etl/LogService.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
37
38

package com.zdjizhi.etl;

import com.zdjizhi.etl.connection.ArangodbBatchIPWindow;
import com.zdjizhi.utils.arangodb.ArangoDBSink;
import com.zdjizhi.utils.ck.ClickhouseSink;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Map;

import static com.zdjizhi.common.FlowWriteConfig.*;

public interface LogService {

    public static void getLogCKSink(DataStream<Map<String, Object>> sourceStream, String sink) throws Exception{

        sourceStream.windowAll(TumblingProcessingTimeWindows.of(Time.milliseconds(SINK_CK_BATCH_DELAY_TIME)))
                .trigger(new CountTriggerWithTimeout<>(sink, CK_BATCH, TimeCharacteristic.ProcessingTime))
                .apply(new CKBatchWindow())
                .addSink(new ClickhouseSink(sink))
                .setParallelism(SINK_PARALLELISM)
                .name(sink)
                .setParallelism(SINK_PARALLELISM);

    }

    public static void getLogArangoSink(DataStream<Map<String, Object>> sourceStream, String sink) throws Exception{
        sourceStream.windowAll(TumblingProcessingTimeWindows.of(Time.milliseconds(SINK_ARANGODB_BATCH_DELAY_TIME)))
                .trigger(new CountTriggerWithTimeout<>(sink, ARANGODB_BATCH, TimeCharacteristic.ProcessingTime))
                .apply(new ArangodbBatchIPWindow())
                .addSink(new ArangoDBSink(sink))
                .setParallelism(SINK_PARALLELISM)
                .name(sink)
                .setParallelism(SINK_PARALLELISM);
    }
}