Moved Flink related code from libraries to libraries-data module. (#5253)

2018-09-15 13:14:57 +02:00
parent 33aa52ac94
commit 05e1700fb7
17 changed files with 71 additions and 49 deletions
@@ -0,0 +1,82 @@
+package com.baeldung.flink;
+
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.model.InputMessage;
+import com.baeldung.flink.operator.BackupAggregator;
+import com.baeldung.flink.operator.InputMessageTimestampAssigner;
+import com.baeldung.flink.operator.WordsCapitalizer;
+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.windowing.time.Time;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
+
+import static com.baeldung.flink.connector.Consumers.*;
+import static com.baeldung.flink.connector.Producers.*;
+
+public class FlinkDataPipeline {
+
+    public static void capitalize() throws Exception {
+        String inputTopic = "flink_input";
+        String outputTopic = "flink_output";
+        String consumerGroup = "baeldung";
+        String address = "localhost:9092";
+
+        StreamExecutionEnvironment environment =
+          StreamExecutionEnvironment.getExecutionEnvironment();
+
+        FlinkKafkaConsumer011<String> flinkKafkaConsumer =
+          createStringConsumerForTopic(inputTopic, address, consumerGroup);
+        flinkKafkaConsumer.setStartFromEarliest();
+
+        DataStream<String> stringInputStream =
+          environment.addSource(flinkKafkaConsumer);
+
+        FlinkKafkaProducer011<String> flinkKafkaProducer =
+        createStringProducer(outputTopic, address);
+
+        stringInputStream
+          .map(new WordsCapitalizer())
+          .addSink(flinkKafkaProducer);
+
+        environment.execute();
+    }
+
+public static void createBackup () throws Exception {
+    String inputTopic = "flink_input";
+    String outputTopic = "flink_output";
+    String consumerGroup = "baeldung";
+    String kafkaAddress = "localhost:9092";
+
+    StreamExecutionEnvironment environment =
+      StreamExecutionEnvironment.getExecutionEnvironment();
+
+    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+
+    FlinkKafkaConsumer011<InputMessage> flinkKafkaConsumer =
+      createInputMessageConsumer(inputTopic, kafkaAddress, consumerGroup);
+    flinkKafkaConsumer.setStartFromEarliest();
+
+    flinkKafkaConsumer
+      .assignTimestampsAndWatermarks(new InputMessageTimestampAssigner());
+    FlinkKafkaProducer011<Backup> flinkKafkaProducer =
+      createBackupProducer(outputTopic, kafkaAddress);
+
+    DataStream<InputMessage> inputMessagesStream =
+      environment.addSource(flinkKafkaConsumer);
+
+    inputMessagesStream
+      .timeWindowAll(Time.hours(24))
+      .aggregate(new BackupAggregator())
+      .addSink(flinkKafkaProducer);
+
+    environment.execute();
+}
+
+    public static void main(String[] args) throws Exception {
+        createBackup();
+    }
+
+}
@@ -0,0 +1,18 @@
+package com.baeldung.flink;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.util.Collector;
+
+import java.util.stream.Stream;
+
+@SuppressWarnings("serial")
+public class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
+
+    @Override
+    public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
+
+        String[] tokens = value.toLowerCase().split("\\W+");
+        Stream.of(tokens).filter(t -> t.length() > 0).forEach(token -> out.collect(new Tuple2<>(token, 1)));
+    }
+}
@@ -0,0 +1,18 @@
+package com.baeldung.flink;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.aggregation.Aggregations;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+import java.util.List;
+
+public class WordCount {
+
+    public static DataSet<Tuple2<String, Integer>> startWordCount(ExecutionEnvironment env, List<String> lines) throws Exception {
+        DataSet<String> text = env.fromCollection(lines);
+
+        return text.flatMap(new LineSplitter()).groupBy(0).aggregate(Aggregations.SUM, 1);
+
+    }
+}
@@ -0,0 +1,32 @@
+package com.baeldung.flink.connector;
+
+import com.baeldung.flink.model.InputMessage;
+import com.baeldung.flink.schema.InputMessageDeserializationSchema;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
+
+import java.util.Properties;
+
+public class Consumers {
+
+public static FlinkKafkaConsumer011<String> createStringConsumerForTopic(
+  String topic, String kafkaAddress, String kafkaGroup ) {
+    Properties props = new Properties();
+    props.setProperty("bootstrap.servers", kafkaAddress);
+    props.setProperty("group.id",kafkaGroup);
+    FlinkKafkaConsumer011<String> consumer =
+      new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(),props);
+
+    return consumer;
+}
+
+    public static FlinkKafkaConsumer011<InputMessage> createInputMessageConsumer(String topic, String kafkaAddress, String kafkaGroup ) {
+        Properties properties = new Properties();
+        properties.setProperty("bootstrap.servers", kafkaAddress);
+        properties.setProperty("group.id",kafkaGroup);
+        FlinkKafkaConsumer011<InputMessage> consumer = new FlinkKafkaConsumer011<InputMessage>(
+                topic, new InputMessageDeserializationSchema(),properties);
+
+        return consumer;
+    }
+}
@@ -0,0 +1,17 @@
+package com.baeldung.flink.connector;
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.schema.BackupSerializationSchema;
+import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
+
+public class Producers {
+
+    public static FlinkKafkaProducer011<String> createStringProducer(String topic, String kafkaAddress) {
+        return new FlinkKafkaProducer011<>(kafkaAddress, topic, new SimpleStringSchema());
+    }
+
+    public static FlinkKafkaProducer011<Backup> createBackupProducer(String topic, String kafkaAddress) {
+        return new FlinkKafkaProducer011<Backup>(kafkaAddress, topic, new BackupSerializationSchema());
+    }
+}
@@ -0,0 +1,27 @@
+package com.baeldung.flink.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.UUID;
+
+public class Backup {
+
+    @JsonProperty("inputMessages")
+    List<InputMessage> inputMessages;
+    @JsonProperty("backupTimestamp")
+    LocalDateTime backupTimestamp;
+    @JsonProperty("uuid")
+    UUID uuid;
+
+    public Backup(List<InputMessage> inputMessages, LocalDateTime backupTimestamp) {
+        this.inputMessages = inputMessages;
+        this.backupTimestamp = backupTimestamp;
+        this.uuid = UUID.randomUUID();
+    }
+
+    public List<InputMessage> getInputMessages() {
+        return inputMessages;
+    }
+}
@@ -0,0 +1,71 @@
+package com.baeldung.flink.model;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.google.common.base.Objects;
+
+import java.time.LocalDateTime;
+
+@JsonSerialize
+public class InputMessage {
+    String sender;
+    String recipient;
+    LocalDateTime sentAt;
+    String message;
+
+    public InputMessage() {
+    }
+
+    public String getSender() {
+        return sender;
+    }
+    public void setSender(String sender) {
+        this.sender = sender;
+    }
+
+    public String getRecipient() {
+        return recipient;
+    }
+
+    public void setRecipient(String recipient) {
+        this.recipient = recipient;
+    }
+
+    public LocalDateTime getSentAt() {
+        return sentAt;
+    }
+
+    public void setSentAt(LocalDateTime sentAt) {
+        this.sentAt = sentAt;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    public InputMessage(String sender, String recipient, LocalDateTime sentAt, String message) {
+        this.sender = sender;
+        this.recipient = recipient;
+        this.sentAt = sentAt;
+        this.message = message;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        InputMessage message1 = (InputMessage) o;
+        return Objects.equal(sender, message1.sender) &&
+                Objects.equal(recipient, message1.recipient) &&
+                Objects.equal(sentAt, message1.sentAt) &&
+                Objects.equal(message, message1.message);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(sender, recipient, sentAt, message);
+    }
+}
@@ -0,0 +1,34 @@
+package com.baeldung.flink.operator;
+
+import com.baeldung.flink.model.Backup;
+import com.baeldung.flink.model.InputMessage;
+import org.apache.flink.api.common.functions.AggregateFunction;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+
+    public class BackupAggregator implements AggregateFunction<InputMessage, List<InputMessage>, Backup> {
+        @Override
+        public List<InputMessage> createAccumulator() {
+            return new ArrayList<>();
+        }
+
+        @Override
+        public List<InputMessage> add(InputMessage inputMessage, List<InputMessage> inputMessages) {
+            inputMessages.add(inputMessage);
+            return inputMessages;
+        }
+
+        @Override
+        public Backup getResult(List<InputMessage> inputMessages) {
+            Backup backup = new Backup(inputMessages, LocalDateTime.now());
+                return backup;
+        }
+
+        @Override
+        public List<InputMessage> merge(List<InputMessage> inputMessages, List<InputMessage> acc1) {
+            inputMessages.addAll(acc1);
+            return inputMessages;
+        }
+    }
@@ -0,0 +1,23 @@
+package com.baeldung.flink.operator;
+
+import com.baeldung.flink.model.InputMessage;
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
+import org.apache.flink.streaming.api.watermark.Watermark;
+
+import javax.annotation.Nullable;
+import java.time.ZoneId;
+
+public class InputMessageTimestampAssigner implements AssignerWithPunctuatedWatermarks<InputMessage> {
+
+    @Override
+    public long extractTimestamp(InputMessage element, long previousElementTimestamp) {
+        ZoneId zoneId = ZoneId.systemDefault();
+        return element.getSentAt().atZone(zoneId).toEpochSecond() * 1000;
+    }
+
+    @Nullable
+    @Override
+    public Watermark checkAndGetNextWatermark(InputMessage lastElement, long extractedTimestamp) {
+        return new Watermark(extractedTimestamp - 15);
+    }
+}
@@ -0,0 +1,11 @@
+package com.baeldung.flink.operator;
+
+import org.apache.flink.api.common.functions.MapFunction;
+
+public class WordsCapitalizer implements MapFunction<String, String> {
+
+    @Override
+    public String map(String s) {
+        return s.toUpperCase();
+    }
+}
@@ -0,0 +1,33 @@
+package com.baeldung.flink.schema;
+
+import com.baeldung.flink.model.Backup;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BackupSerializationSchema
+  implements SerializationSchema<Backup> {
+
+    static ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+
+    Logger logger = LoggerFactory.getLogger(BackupSerializationSchema.class);
+
+    @Override
+    public byte[] serialize(Backup backupMessage) {
+        if(objectMapper == null) {
+            objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY);
+            objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+        }
+        try {
+            String json = objectMapper.writeValueAsString(backupMessage);
+            return json.getBytes();
+        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
+            logger.error("Failed to parse JSON", e);
+        }
+        return new byte[0];
+    }
+}
@@ -0,0 +1,32 @@
+package com.baeldung.flink.schema;
+
+import com.baeldung.flink.model.InputMessage;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+import java.io.IOException;
+
+public class InputMessageDeserializationSchema implements
+      DeserializationSchema<InputMessage> {
+
+    static ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+
+
+    @Override
+    public InputMessage deserialize(byte[] bytes) throws IOException {
+
+        return objectMapper.readValue(bytes, InputMessage.class);
+    }
+
+    @Override
+    public boolean isEndOfStream(InputMessage inputMessage) {
+        return false;
+    }
+
+    @Override
+    public TypeInformation<InputMessage> getProducedType() {
+        return TypeInformation.of(InputMessage.class);
+    }
+}