1. DataStream 简介

Flink 中的 DataStream 程序是对数据流（例如过滤、更新状态、定义窗口、聚合）进行转换的常规程序。数据流的起始是从各种源（例如消息队列、套接字流、文件）创建的。结果通过 sink 返回，例如可以将数据写入文件或标准输出（例如命令行终端）。Flink 程序可以在各种上下文中运行，可以独立运行，也可以嵌入到其它程序中。任务执行可以运行在本地 JVM 中，也可以运行在多台机器的集群上。

2. Flink 程序

每个Flink程序由相同的基本部分组成，程序结构大致如下：

  1. 获取一个执行环境；
  2. 从指定数据源加载初始数据（Source）；
  3. 指定数据相关的转换（Transform）；
  4. 将计算结果输出到目标存储（Sink）；
  5. 触发程序执行。

image.png

2.1 getExecutionEnvironment

创建一个执行环境，表示当前执行程序的上下文。 如果程序是独立调用的，则此方法返回本地执行环境；如果从命令行客户端调用程序以提交到集群，则此方法返回此集群的执行环境，也就是说， getExecutionEnvironment会根据查询运行的方式决定返回什么样的运行环境，是最常用的一种创建执行环境的方式。

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

2.2 createLocalEnvironment

返回本地执行环境。

  1. LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

2.3 createRemoteEnvironment

返回集群执行环境，将Jar提交到远程服务器。需要在调用时指定JobManager的IP和端口号，并指定要在集群中运行的Jar包。

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(String host, int port, String... jarFiles);

3. Maven配置

亲测可用！！！

akka是用scala实现的，所以这里用到了scala实现的包

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  7.     <groupId>org.example</groupId>
  8.     <artifactId>blink</artifactId>
  9.     <version>1.0-SNAPSHOT</version>
  11.     <properties>
  12.         <java.version>1.8</java.version>
  13.         <fastjson.version>1.2.75</fastjson.version>
  14.         <druid.version>1.2.5</druid.version>
  15.         <flink.version>1.13.1</flink.version>
  16.         <scala.binary.version>2.12</scala.binary.version>
  17.         <HikariCP.version>3.2.0</HikariCP.version>
  18.         <Impala.version>2.6.4</Impala.version>
  19.         <kafka.version>2.8.0</kafka.version>
  20.     </properties>
  21.     <dependencies>
  22.         <dependency>
  23.             <groupId>org.apache.flink</groupId>
  24.             <artifactId>flink-java</artifactId>
  25.             <version>${flink.version}</version>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.apache.flink</groupId>
  29.             <artifactId>flink-streaming-java_2.11</artifactId>
  30.             <version>${flink.version}</version>
  31.         </dependency>
  32.         <dependency>
  33.             <groupId>org.apache.flink</groupId>
  34.             <artifactId>flink-scala_${scala.binary.version}</artifactId>
  35.             <version>${flink.version}</version>
  36.         </dependency>
  37.         <dependency>
  38.             <groupId>org.apache.flink</groupId>
  39.             <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
  40.             <version>${flink.version}</version>
  41.         </dependency>
  42.         <dependency>
  43.             <groupId>org.apache.flink</groupId>
  44.             <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
  45.             <version>${flink.version}</version>
  46.         </dependency>
  47.         <dependency>
  48.             <groupId>org.apache.flink</groupId>
  49.             <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
  50.             <version>${flink.version}</version>
  51.         </dependency>
  52.         <dependency>
  53.             <groupId>org.apache.flink</groupId>
  54.             <artifactId>flink-table-common</artifactId>
  55.             <version>${flink.version}</version>
  56.         </dependency>
  57.         <dependency>
  58.             <groupId>org.apache.flink</groupId>
  59.             <artifactId>flink-clients_${scala.binary.version}</artifactId>
  60.             <version>${flink.version}</version>
  61.         </dependency>
  62.         <dependency>
  63.             <groupId>org.apache.flink</groupId>
  64.             <artifactId>flink-clients_2.11</artifactId>
  65.             <version>${flink.version}</version>
  66.         </dependency>
  67.         <dependency>
  68.             <groupId>org.apache.flink</groupId>
  69.             <artifactId>flink-table-planner-blink_2.12</artifactId>
  70.             <version>${flink.version}</version>
  71.         </dependency>
  72.         <dependency>
  73.             <groupId>org.apache.flink</groupId>
  74.             <artifactId>flink-json</artifactId>
  75.             <version>${flink.version}</version>
  76.         </dependency>
  77.         <dependency>
  78.             <groupId>com.alibaba.ververica</groupId>
  79.             <artifactId>flink-connector-mysql-cdc</artifactId>
  80.             <version>1.4.0</version>
  81.         </dependency>
  82.         <dependency>
  83.             <groupId>org.apache.flink</groupId>
  84.             <artifactId>flink-runtime_2.11</artifactId>
  85.             <version>${flink.version}</version>
  86.         </dependency>
  87.         <dependency>
  88.             <groupId>org.apache.flink</groupId>
  89.             <artifactId>flink-connector-kafka_2.11</artifactId>
  90.             <version>${flink.version}</version>
  91.         </dependency>
  92.         <dependency>
  93.             <groupId>org.apache.flink</groupId>
  94.             <artifactId>flink-sql-connector-kafka_2.11</artifactId>
  95.             <version>${flink.version}</version>
  96.         </dependency>
  97.         <dependency>
  98.             <groupId>org.apache.flink</groupId>
  99.             <artifactId>flink-connector-jdbc_2.11</artifactId>
  100.             <version>${flink.version}</version>
  101.         </dependency>
  102.         <dependency>
  103.             <groupId>com.zaxxer</groupId>
  104.             <artifactId>HikariCP</artifactId>
  105.             <version>${HikariCP.version}</version>
  106.         </dependency>
  107.         <dependency>
  108.             <groupId>mysql</groupId>
  109.             <artifactId>mysql-connector-java</artifactId>
  110.             <version>8.0.13</version>
  111.         </dependency>
  112.         <dependency>
  113.             <groupId>org.slf4j</groupId>
  114.             <artifactId>slf4j-api</artifactId>
  115.             <version>1.7.25</version>
  116.         </dependency>
  117.         <dependency>
  118.             <groupId>org.slf4j</groupId>
  119.             <artifactId>slf4j-log4j12</artifactId>
  120.             <version>1.7.25</version>
  121.         </dependency>
  122.         <dependency>
  123.             <groupId>com.google.code.gson</groupId>
  124.             <artifactId>gson</artifactId>
  125.             <version>2.8.2</version>
  126.         </dependency>
  127.     </dependencies>
  128.     <build>
  129.         <plugins>
  130.             <plugin>
  131.                 <groupId>org.apache.maven.plugins</groupId>
  132.                 <artifactId>maven-assembly-plugin</artifactId>
  133.                 <version>3.1.0</version>
  134.                 <configuration>
  135.                     <descriptorRefs>
  136.                         <descriptorRef>jar-with-dependencies</descriptorRef>
  137.                     </descriptorRefs>
  138.                 </configuration>
  139.                 <executions>
  140.                     <execution>
  141.                         <id>make-assembly</id>
  142.                         <phase>package</phase>
  143.                         <goals>
  144.                             <goal>single</goal>
  145.                         </goals>
  146.                     </execution>
  147.                 </executions>
  148.             </plugin>
  149.             <plugin>
  150.                 <groupId>org.apache.maven.plugins</groupId>
  151.                 <artifactId>maven-compiler-plugin</artifactId>
  152.                 <configuration>
  153.                     <source>8</source>
  154.                     <target>8</target>
  155.                 </configuration>
  156.             </plugin>
  157.         </plugins>
  158.     </build>
  159. </project>

4. 示例程序

  1. package com.nkong.blink.start;
  3. import org.apache.flink.api.common.functions.FlatMapFunction;
  4. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  5. import org.apache.flink.api.java.functions.KeySelector;
  6. import org.apache.flink.api.java.tuple.Tuple2;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
  10. import org.apache.flink.streaming.api.windowing.time.Time;
  11. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  12. import org.apache.flink.util.Collector;
  14. import java.util.Properties;
  16. /**
  17.  * 每5s统计一次窗口元素个数
  18.  * @author nkong
  19.  * @time 2022/01/25 14:38
  20.  */
  21. public class SocketInputJob {
  23.     public static void main(String[] args) throws Exception {
  24.         // 获取Flink执行环境
  25.         // 并行度：默认是计算机的CPU逻辑核数
  26.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         // 从指定Socket读取数据
  28.         DataStream<String> text = env.socketTextStream("127.0.0.1", 9000);
  29.         DataStream<Tuple2<String, Integer>> dataStream = text.flatMap(new MyFlatMap())
  30.                 .keyBy(new MyKeySelector())
  31.                 // 滚动时间窗口
  32.                 .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
  33.                 .sum(1);
  34.         dataStream.print();
  35.         env.execute("SocketInputJob");
  36.     }
  38.     private static class MyFlatMap implements FlatMapFunction<String, Tuple2<String, Integer>> {
  39.         @Override
  40.         public void flatMap(String inStr, Collector<Tuple2<String, Integer>> collector) throws Exception {
  41.             String[] tokens = inStr.toLowerCase().split("\t", -1);
  42.             for (String token : tokens) {
  43.                 if (token.length() > 0) {
  44.                     collector.collect(new Tuple2<String, Integer>(token, 1));
  45.                 }
  46.             }
  47.         }
  48.     }
  50.     private static class MyKeySelector implements KeySelector<Tuple2<String, Integer>, Object> {
  51.         @Override
  52.         public Object getKey(Tuple2<String, Integer> stringIntegerTuple) throws Exception {
  53.             return stringIntegerTuple.f0;
  54.         }
  55.     }
  57. }