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

package com.zdjizhi.etl;

import cn.hutool.json.JSONUtil;
import com.zdjizhi.utils.ck.CKSink;
import com.zdjizhi.utils.kafka.KafkaProducer;
import org.apache.flink.streaming.api.datastream.DataStream;

import java.util.Map;

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

public class LogService {

    public static void getLogKafkaSink(DataStream<Map<String, Object>> sourceStream, String sink) throws Exception {
        sourceStream.map(JSONUtil::toJsonStr)
                .setParallelism(SINK_PARALLELISM)
                .addSink(KafkaProducer.getKafkaProducer(sink))
                .setParallelism(SINK_PARALLELISM)
                .name(sink)
                .setParallelism(SINK_PARALLELISM);
    }

    public static void getLogCKSink(DataStream<Map<String, Object>> sourceStream, String sink) throws Exception {
        sourceStream.addSink(new CKSink(sink))
                .setParallelism(SINK_PARALLELISM)
                .name(sink)
                .setParallelism(SINK_PARALLELISM);

    }
}