1.入门 demo

WordCount.java

  1. public class WordCount {
  2.     public static void main(String[] args) throws Exception {
  3.         //步骤一：获取执行环境
  4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  5.         //步骤二：获取数据源
  7.         DataStreamSource<String> dataStream = env.socketTextStream("localhost", 8888);
  8.         //步骤三：数据处理
  9.         SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
  10.             @Override
  11.             public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
  12.                 String[] fields = line.split(",");
  13.                 for (String word : fields) {
  14.                     collector.collect(new Tuple2<>(word, 1));
  15.                 }
  16.             }
  18.         });
  20.         SingleOutputStreamOperator<Tuple2<String, Integer>> wordCount = wordAndOne.keyBy(0).sum(1);
  21.         wordCount.print();
  22.         //步骤四：数据输出 wordCount.print(); //步骤五：启动任务
  23.         env.execute("word count ...");
  25.     }
  26. }

pom.xml

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.apache.flink</groupId>
  4.         <artifactId>flink-streaming-java_2.11</artifactId>
  5.         <version>${flink.version}</version>
  6.     </dependency>
  7. </dependencies>

2.本地调试

  1. nc -lk 8888
  2. a,a,b,c

本地可视化界面

1.增加依赖

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-runtime-web_2.11</artifactId>
  4.     <version>${flink.version}</version>
  5. </dependency>

2.修改 env 的获取方式

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

3.浏览器启动
http://localhost:8081/

Flink(1) 编程模型 - 图1

3.集群运行

1.打包

pom.xml

  1.     <build>
  2.         <plugins>
  3.             <plugin>
  4.                 <groupId>org.apache.maven.plugins</groupId>
  5.                 <artifactId>maven-shade-plugin</artifactId>
  6.                 <version>3.2.1</version>
  7.                 <executions>
  8.                     <execution>
  9.                         <phase>package</phase>
  10.                         <goals>
  11.                             <goal>shade</goal>
  12.                         </goals>
  13.                         <configuration>
  14.                             <transformers>
  15.                                 <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  16.                                     <mainClass>sparkstreaming.wordcount.WordCountStreaming</mainClass>
  17.                                 </transformer>
  18.                             </transformers>
  19.                         </configuration>
  20.                     </execution>
  21.                 </executions>
  22.             </plugin>
  23.         </plugins>
  24.     </build>

mvn clean package

得到 learn_bigdata_flink-1.0-SNAPSHOT.jar

2.提交任务

  1. flink run --class wc.wordcount --jar learn_bigdata_flink-1.0-SNAPSHOT.jar

2.算子

DataSet

算子 含义
Map 输入一个元素，然后返回一个元素，中间可以做一些清洗转换等操作
FlatMap 输入一个元素，可以返回零个，一个或者多个元素
MapPartition 类似map，一次处理一个分区的数据【如果在进行map处理的时候需要获取第三方资源 链接，建议使用MapPartition】
Filter 过滤函数，对传入的数据进行判断，符合条件的数据会被留下
Reduce 对数据进行聚合操作，结合当前元素和上一次reduce返回的值进行聚合操作，然后返回一个 新的值
Aggregate sum、max、min等
Distinct 返回一个数据集中去重之后的元素，data.distinct()
Join 内连接
OuterJoin 外链接
Cross 获取两个数据集的笛卡尔积
Union 返回两个数据集的总和，数据类型需要一致
First-n 获取集合中的前N个元素
SortPartition 在本地对数据集的所有分区进行排序，通过sortPartition()的链接调用来完成对多个字段的排序

广播变量

广播变量允许编程人员在每台机器上保持1个只读的缓存变量，而不是传送变量的副本给tasks 广播变量创建后，它可以运行在集群中的任何function上，而不需要多次传递给集群节点。另外需要记 住，不应该修改广播变量，这样才能确保每个节点获取到的值都是一致的 一句话解释，可以理解为是一个公共的共享变量，我们可以把一个dataset 数据集广播出去，然后不同 的task在节点上都能够获取到，这个数据在每个节点上只会存在一份。如果不使用broadcast，则在每 个节点中的每个task中都需要拷贝一份dataset数据集，比较浪费内存(也就是一个节点中可能会存在多份dataset数据)。 用法

  1. 初始化数据
    DataSet toBroadcast = env.fromElements(1, 2, 3)
  2. 广播数据
    withBroadcastSet(toBroadcast, “broadcastSetName”);
  3. 获取数据
    Collection broadcastSet =
    getRuntimeContext().getBroadcastVariable(“broadcastSetName”);

计数器

Accumulator即累加器，与Mapreduce counter的应用场景差不多，都能很好地观察task在运行期间的数 据变化 可以在Flink job任务中的算子函数中操作累加器，但是只能在任务执行结束之后才能获得累加器的最终结果。 Counter是一个具体的累加器(Accumulator)实现 IntCounter, LongCounter 和 DoubleCounter 用法

  1. 创建累加器
    private IntCounter numLines = new IntCounter();
  2. 注册累加器
    getRuntimeContext().addAccumulator(“num-lines”, this.numLines);
  3. 使用累加器
    this.numLines.add(1);
  4. 获取累加器的结果
    myJobExecutionResult.getAccumulatorResult(“num-lines”)