Apache Flink 학습 노트 (2)

이전 Apache Flink 학습 노트 (1) 는 일괄 처리 의 사용 을 간단하게 시범 하 였 으 며, 이 편 은 스 트림 처리 의 사용 방법 을 보 여 주 었 다.
흐름 처리 도 무한 처리 라 고 합 니 다. 데 이 터 는 끊임없이 로드 되 기 때문에 흐름 처 리 는 DataStream 류 를 사용 해 야 합 니 다.본 편 demo 은 kafka (회사 에 이미 만들어 진 소식 생산자 가 있다) 과 결합 하여 시연 할 것 이다.kafka 메시지 체 는 다음 과 같다 (json).

{
    "appId":"xxxx",
    "module":"xxxx"
    //    
}

지금 은 10s 한 번 씩 집계 하고 싶 습 니 다. appid 조별 계수 (간단 한 수요) 에 따라 Event Time ProcessingTime, Windows 입 니 다.

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

public class Demo3 {
    public static void main(String[] args) {
        //           StreamExecutionEnvironment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().enableSysoutLogging();//  Sysout   
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); //          process time
        env.setParallelism(2);//     
        //  kafka bootstrap.servers
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "kafka bootstrap.servers");
        //          （      FlinkKafkaManager，     ）
        FlinkKafkaManager manager = new FlinkKafkaManager("kafka.topic", "app.name", properties);
        // JsonObject       kafka
        FlinkKafkaConsumer09 consumer = manager.build(JSONObject.class);
        //          
        consumer.setStartFromLatest();
        //  DataStream
        DataStream messageStream = env.addSource(consumer);
        //   pojo
        DataStream bean3DataStream = messageStream.map(new FlatMap());
        bean3DataStream
                .keyBy(Bean3::getAppId) //    “appId”  
                .timeWindow(Time.seconds(10))//        ，       TimeCharacteristic.ProcessingTime
               // .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))//  process time   
                .aggregate(new Agg()) //    ，       demo2 reduce  
                .addSink(new Sink()); //    
        try {
            env.execute("app.name");//          
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class FlatMap implements MapFunction {
        @Override
        public Bean3 map(JSONObject jsonObject) throws Exception {
            return new Bean3(jsonObject.getString("appId"), jsonObject.getString("module"));
        }
    }

    public static class Agg implements AggregateFunction, Tuple2> {
        @Override
        public Tuple2 createAccumulator() {
            return new Tuple2();
        }

        @Override
        public Tuple2 add(Bean3 bean3, Tuple2 bean3LongTuple2) {
            Bean3 bean = bean3LongTuple2.f0;
            Long count = bean3LongTuple2.f1;
            if (bean == null) {
                bean = bean3;
            }
            if (count == null) {
                count = 1L;
            } else {
                count++;
            }
            return new Tuple2<>(bean, count);
        }

        @Override
        public Tuple2 getResult(Tuple2 bean3LongTuple2) {
            return bean3LongTuple2;
        }

        @Override
        public Tuple2 merge(Tuple2 bean3LongTuple2, Tuple2 acc1) {
            Bean3 bean = bean3LongTuple2.f0;
            Long count = bean3LongTuple2.f1;
            Long acc = acc1.f1;
            return new Tuple2<>(bean, count + acc);
        }
    }

    public static class Sink implements SinkFunction> {
        @Override
        public void invoke(Tuple2 value, Context context) throws Exception {
            System.out.println(value.f0.toString() + "," + value.f1);
        }
    }

    public static class Bean3 {
        public String appId;
        public String module;

        public Bean3() {
        }

        public Bean3(String appId, String module) {
            this.appId = appId;
            this.module = module;
        }

        public String getAppId() {
            return appId;
        }

        public void setAppId(String appId) {
            this.appId = appId;
        }

        public String getModule() {
            return module;
        }

        public void setModule(String module) {
            this.module = module;
        }

        @Override
        public String toString() {
            return "Bean3{" +
                    "appId='" + appId + '\'' +
                    ", module='" + module + '\'' +
                    '}';
        }
    }
}

이전 편 에서 일괄 처리 한 demo 에 비해 흐름 처 리 는 매우 복잡 해 보인다.실제로 두 사람 은 납득 할 만 한 부분 이 많다. 예 를 들 어 일괄 처리 중의 groupBy 과 흐름 처리 keyBy 는 모두 지 정 된 차원 에 따라 조 를 나 누 었 다.
한편, 스 트림 처리 에서 의 개념 을 도입 할 것 이다. 앞에서 말 한 바 와 같이 스 트림 데 이 터 는 무한 데이터 이 고 Flink 창 을 통 해 무한 데 이 터 를 하나의 '일괄 처리' 로 전환 시 켜 계산 할 것 이다.창 구 는 , , 등 으로 나 뉘 는데 구체 적 으로 홈 페이지 소 개 를 참조 할 수 있다.각 창의 시간 구분 은 event time 에 의 해 결정 되 는데 본 사례 는 ProcessingTime 즉 처리 시간 을 사용한다.
다음은 demo3 을 개조 하여 사용 EventTime 으로 만 들 겠 습 니 다. 즉, 창의 시간 은 데이터 소스 의 시간 스탬프 (사건 발생) 에 의 해 결정 된다 는 것 입 니 다.
바꾸다

// pojo Bean3        
public static class Bean3 {
    public Long timestamp;//add event time
    public String appId;
    public String module;

    public Bean3() {
    }

    public Bean3(Long timestamp, String appId, String module) {
        this.timestamp = timestamp;
        this.appId = appId;
        this.module = module;
    }

    public long getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(Long timestamp) {
        this.timestamp = timestamp;
    }
//    
}

고치다

//          event time  
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); 

바꾸다

//  
//         ，Time.seconds(int)            
DataStream bean3DataStreamWithAssignTime = 
bean3DataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor(Time.seconds(0)) {
    @Override
    public long extractTimestamp(Bean3 element) {
        return element.getTimestamp();
    }
});

바꾸다

bean3DataStreamWithAssignTime
                .keyBy(Bean3::getAppId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))//  event time   
                .allowedLateness(Time.seconds(5)) //          ,    ,            
//    

FlinkKafkaManager 소스 코드

package flink.test.manager;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import java.util.Properties;

public class FlinkKafkaManager {
    private String topic;
    private String groupId;
    private Properties properties;

    public FlinkKafkaManager(String topic, String groupId, Properties properties) {
        this.topic = topic;
        this.groupId = groupId;
        this.properties = properties;
        this.properties.setProperty("group.id", this.groupId);
        //     kafka         
        this.setDefaultKafkaProperties();
    }

    private void setDefaultKafkaProperties() {
        //  auto commit offset,  5s commit  
        this.properties.setProperty("enable.auto.commit", "true");
        this.properties.setProperty("auto.commit.interval.ms", "5000");
    }

    public FlinkKafkaConsumer09 build(Class clazz) {
        if (checkProperties()) {
            return new FlinkKafkaConsumer09(topic, new ConsumerDeserializationSchema(clazz), properties);
        } else {
            return null;
        }
    }

    private boolean checkProperties() {
        boolean isValued = true;

        if (!properties.containsKey("bootstrap.servers")) {
            isValued = false;
        } else {
            String brokers = properties.getProperty("bootstrap.servers");
            if (brokers == null || brokers.isEmpty()) {
                isValued = false;
            }
        }

        if (this.topic == null || this.topic.isEmpty()) {
            isValued = false;
        }

        if (!properties.containsKey("group.id")) {
            isValued = false;
        } else {
            String groupId = properties.getProperty("group.id");
            if (groupId == null || groupId.isEmpty()) {
                isValued = false;
            }
        }

        return isValued;
    }
}

소비자 DeserializationSchema 소스 코드

package flink.test.manager;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;

import java.io.IOException;

public class ConsumerDeserializationSchema implements DeserializationSchema {
    private Class clazz;

    public ConsumerDeserializationSchema(Class clazz) {
        this.clazz = clazz;
    }

    @Override
    public T deserialize(byte[] bytes) throws IOException {
        //   new String(bytes)  json   ，    ，     
        return JSON.parseObject(new String(bytes), clazz);
    }

    @Override
    public boolean isEndOfStream(T t) {
        return false;
    }

    @Override
    public TypeInformation getProducedType() {
        return TypeExtractor.getForClass(clazz);
    }
}