1.简单讲解

对于这个简短的程序，逻辑还是非常简单的。

  1. 首先创建一个流处理环境env，
  2. 然后往这个环境添加数据源，比如env.socketTextStream，和env.addSource(kafkaSource);
  3. 自定义算子，算子的作用是对传输过来的每一条数据进行处理，是数据处理的核心部分。在下面的程序里，我们重写了flatMap()方法，flatMap+keyBy+sum就完成了一条消息的word count。

  4. env.execute()执行任务。实际上前面的部分是我们定义了这个任务的执行规则，到这一行才开始执行任务，按上述规则对每个事件（指收到消息）进行处理。

    2.maven依赖

    1. <dependencies>
    2.   <dependency>
    3.       <groupId>org.slf4j</groupId>
    4.       <artifactId>slf4j-log4j12</artifactId>
    5.       <version>1.7.7</version>
    6.       <scope>runtime</scope>
    7.   </dependency>
    8.   <dependency>
    9.       <groupId>log4j</groupId>
    10.       <artifactId>log4j</artifactId>
    11.       <version>1.2.17</version>
    12.       <scope>runtime</scope>
    13.   </dependency>
    15.   <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
    16.   <dependency>
    17.       <groupId>org.apache.kafka</groupId>
    18.       <artifactId>kafka_2.11</artifactId>
    19.       <version>2.3.1</version>
    20.   </dependency>
    21.   <dependency>
    22.       <groupId>org.apache.kafka</groupId>
    23.       <artifactId>kafka-clients</artifactId>
    24.       <version>2.3.1</version>
    25.   </dependency>
    27.   <!-- Apache Flink dependencies -->
    28.   <!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
    29.   <dependency>
    30.       <groupId>org.apache.flink</groupId>
    31.       <artifactId>flink-java</artifactId>
    32.       <version>1.10.2</version>
    33.       <scope>provided</scope>
    34.   </dependency>
    36.   <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
    37.   <dependency>
    38.       <groupId>org.apache.flink</groupId>
    39.       <artifactId>flink-streaming-java_2.11</artifactId>
    40.       <version>1.10.2</version>
    41.       <scope>provided</scope>
    42.   </dependency>
    44.   <dependency>
    45.       <groupId>org.apache.flink</groupId>
    46.       <artifactId>flink-connector-kafka-base_2.11</artifactId>
    47.       <version>1.10.2</version>
    48.   </dependency>
    49.   <dependency>
    50.       <groupId>org.apache.flink</groupId>
    51.       <artifactId>flink-connector-kafka_2.11</artifactId>
    52.       <version>1.10.2</version>
    53.   </dependency>

    需要注意的是我们有两处provided，所以要在IDEA->Run->edit configurations里选上 Include provided scope那一行
    image.png

3.Java代码

flinkstream类，接收Kafka生产者发送的消息，对每条消息进行word count

  1. import org.apache.flink.api.common.functions.FlatMapFunction;
  2. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  3. import org.apache.flink.api.java.tuple.Tuple2;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  6. import org.apache.flink.streaming.api.windowing.time.Time;
  7. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  8. import org.apache.flink.util.Collector;
  10. import java.util.Properties;
  12. public class flinkstream {
  13.     public static void main(String[] args) throws Exception {
  14.         //创建流处理环境
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  17.         //接收一个socket文本流，参数：主机名，端口号
  18.         // DataStream<String> text = env.socketTextStream("127.0.0.1", 9000);
  20.         System.out.println("设置kafka连接参数!!!");
  21.         Properties props = new Properties();
  22.         props.setProperty("bootstrap.servers", "172.27.210.57:9092");
  23.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  24.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  25.         props.put("group.id", "cug/kafka/test");
  26.         //如果没有记录偏移量，第一次从最开始消费
  27.         //props.setProperty("auto.offset.reset", "earliest");
  28.         //kafka的消费者不自动提交偏移量
  29.         // props.setProperty("enable.auto.commit","false");
  30.         System.out.println("给flink添加上kafka数据源!!!");
  31.         //kafkaSource,flink消费指定topic的数据
  32.         FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>("lt", new SimpleStringSchema(), props);
  33.         DataStream<String> lines = env.addSource(kafkaSource);
  35.         //进行转换处理
  36.         DataStream<Tuple2<String, Integer>> dataStream = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
  37.             @Override
  38.             public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
  39.                 String[] tokens = s.toLowerCase().split("\\W+");
  41.                 for (String token : tokens) {
  42.                     if (token.length() > 0) {
  43.                         collector.collect(new Tuple2<String, Integer>(token, 1));
  44.                     }
  45.                 }
  46.             }
  47.         }).keyBy(0).timeWindow(Time.seconds(5)).sum(1);
  48.         //keyby是分组 ，sum叠加
  50.         //打印结果
  51.         dataStream.print();
  53.         //启动任务执行，execute可以不给参数，参数是作业名字
  54.         env.execute("Java WordCount from SocketTextStream Example");
  55.     }
  56. }

TestProducer，一个简单的Kafka生产者，发送消息出去

  1. import org.apache.kafka.clients.producer.KafkaProducer;
  2. import org.apache.kafka.clients.producer.ProducerRecord;
  4. import java.util.Properties;
  5. import java.util.Scanner;
  7. public class TestProducer {
  8.     public static void main(String[] args) {
  9.         Properties props = new Properties();
  10.         props.setProperty("bootstrap.servers", "172.27.210.57:9092");
  11.         props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  12.         props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  14.         KafkaProducer<String, String> producer = new KafkaProducer<>(props);
  15.         String topic = "lt";
  17.         while(true){
  18.             System.out.println("请输入一个字符串");
  19.             Scanner sc = new Scanner(System.in);
  20.             String str = sc.nextLine();
  21.             producer.send(new ProducerRecord<>(topic, str));
  23.         }
  24.     }
  25. }

4.运行结果

Kafka发送两条消息
image.png
flink对每条消息的处理结果
image.png
如果不懂Kafka，可以换成netcat发送数据，
18行取消注释
DataStream text = env.socketTextStream(“127.0.0.1”, 9000);
36行lines改成text
DataStream> dataStream = lines.flatMap()
注释调20-33行Kafka那一段
运行netcat，结果如图
image.png
image.png

5.关于重写方法

个人觉得上面那种写法有点不易读，主要是实现了flatMap()里面的接口FlatMapFunction，补全的它的方法flatMap()，可以写一个类来实现它，改动如下：

  1. DataStream<Tuple2<String, Integer>> dataStream = text.flatMap(new Splitter()).keyBy(0).sum(1);
  1.     public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
  2.         @Override
  3.         public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
  4.             String[] tokens = s.toLowerCase().split("\\W+");
  6.             for (String token : tokens) {
  7.                 if (token.length() > 0) {
  8.                     collector.collect(new Tuple2<String, Integer>(token, 1));
  9.                 }
  10.             }
  11.         }
  12.     }

以flatMap对应的FlatMapFunction为例，它在源码中的定义为：

  1. public interface FlatMapFunction<T, O> extends Function, Serializable {
  2.   void flatMap(T value, Collector<O> out) throws Exception;
  4. }

这是一个接口类，它继承了Flink的Function函数式接口。函数式接口只有一个抽象函数方法（Single Abstract Method），其目的是为了方便Java 8 Lambda表达式的使用。此外，它还继承了Serializable，以便进行序列化，这是因为这些函数在运行过程中要发送到各个TaskManager上，发送前后要进行序列化和反序列化。需要注意的是，使用这些函数时，一定要保证函数内的所有内容都可以被序列化。如果有一些不能被序列化的内容，或者使用接下来介绍的Rich函数类，或者重写Java的序列化和反序列化方法。
进一步观察FlatMapFunction发现，这个这个函数有两个泛型T和O，T是输入，O是输出，在使用时，要设置好对应的输入和输出数据类型。自定义函数最终归结为重写函数flatMap，函数的两个参数也与输入输出的泛型类型对应，即参数value的是flatMap的输入，数据类型是T，参数out是flatMap的输出，我们需要将类型为O的数据写入out。
在我们的程序里，泛型O用了Tuple2，Tuple2又是一个泛型类，是flink里定义的，在里面设置参数类型为，就可以做到单词计数了。