Flink入门实战
Flink项目构建
1)基于Maven+Idea创建项目：
使用maven进行项目构建，如图1所示。
图-34 构建maven项目
输入项目中的maven的坐标和存储坐标，如图2所示。
图2 maven坐标和存储位置
2)Maven依赖：
Flink基础API概念
Flink编程是在分布式集合的基础的规律的编程模型（好比，执行filtering,mapping,updating,state,joining,grouping,defining,windows,aggregating）。这些集合可以通过外部数据源（好比从文件，kafka的topics、本地大概内存的集合）。通过下沉算子返回结果，好比将数据写入到一个分布式的文件中，大概控制台。Flink程序可以基于各种context、stanalone大概嵌入其他程序进行运行。可以在本地的jvm大概在多台呆板间分布式运行。
基于外部的数据源，好比有界大概无界的数据源，我们大概会选择使用批处理的DataSet API大概流处理的DataStream API来处理。
需要注意的是，在DataStream和DataSet中的绝大多数的API是同等的，只需要替换对应的ExecutionEnvironment大概StreamExecutionEnvironment即可。
Flink在编程的过程中使用特定类——DataSet和DataStream来表现数据，类似Spark中的RDD。可以将其认为是一个可以拥有重复的不可变的集合。其中DataSet表现的是一个有界的数据集，DataStream则表现的是无界的集合。
这些集合在一些关键的地方和Java中的普通集合不同。首先，DataSet和DataStream是不可变的，这就意味着一旦被创建，便不能进行add大概remove的操纵。同样也不能简单的查看集合内部的元素。
Flink可以通过外部的数据源来创建DataSet大概DataStream，也可以通过在一个已知的集合上面执行一系列的Transformation操纵来转换产生新的集合。
Flink程序看起来就是一个普通的程序，进行数据的转换，每一个程序包罗如下相同的集合基础概念，通用编程步调如下：
1)创建一个执行环境ExecutionEnvironment。
2)加载大概创建初始化数据——DataSet大概DataStream。
3)在此数据基础之上进行特定的转化操纵。
4)将计算的结果输出到特定的目的地。
5)触发作业的执行。
Flink编程的入口，便是ExecutionEnvironment，不同之处在于，DataSet和DataStream使用的ExecutionEnvironment不同。DataSet使用ExecutionEnviroment，而DataStream使用StreamExectionEnvironment。
得到ExecutionEnvironment可以通过ExecutionEnvironment的如下方法：
getExecutionEnvironment()
createLocalEnvironment()
createRemoteEnvironment(String host, int port, String... jarFiles)
通常环境下，我们只需要使用getExecutionEnvironment()即可，因为这种方式会自动选择正确的context。假如我们在IDE中执行，则会创建一个Local的Context，假如打包到集群中执行，会返回一个Cluster的Context。
加载数据源的方式有多种。可以一行一个的读入，好比CSV文件，大概自定义格式。假如只是从一个文本文件中按次序读取行数据。只需要如下操纵即可。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> text = env.readTextFile("file:///path")
创建了一个DataStream大概DataSet，接下来便可以执行各种transformation转换操纵。好比执行一个map操纵。
创建一个新的DataStream，类型为Integer的集合。
DataStream中包罗了最终的结果，我们可以将结果通过创建一个sink操纵，写入外部系统中，好比：writeAsText(path)
一旦我们完成整个程序，我们需要通过调用StreamExecutionEnvironment的execute()方法来触发作业的执行。基于ExecutionEnvironment会在本地大概集群中执行。
execute()方法返回值为JobExecutionResult，包罗本次执行时间大概累加器结果信息。
与Spark中的Transformation操纵相同，Flink中的Transformation操纵是Lazy懒加载的，需要execute()去触发。基于此，我们可以创建并添加程序的执行计划。进行任务调度和数据分离，执行更加高效。
目前Flink支持7种数据类型，分别为：
1)Java Tuples和Scala Case Classes。
2)Java POJOS(一种数据结构类型)。
3)Primitive Types(Java的根本数据类型)。
4)Regular Classes(普通类)。
5)Values。
6)Hadoop Writables。
7)SpecialTypes。
DataSet批处理API
Flink中的DataSet程序是实现数据集转换的常规程序（例如，Filter，映射，连接， 分组）。数据集最初是从某些来源创建的（例如，通过读取文件或从本地集合创建）。结果通过接收器返回，接收器可以例如将数据写入（分布式）文件或标准输出（例如命令行终端）。Flink程序可以在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在很多计算机的集群上执行。
- import org.apache.flink.api.scala._
- object WordCountOps {
- def main(args: Array[String]): Unit = {
- val env = ExecutionEnvironment.getExecutionEnvironment
- val text = env.fromElements("Who's there?",
- "I think I hear them. Stand, ho! Who's there?"
- )
- val wordCounts:DataSet[(String, Int)] = text
- .flatMap(line => line.split("\\s+")).map((_, 1))
- .groupBy(0)
- .sum(1)
- wordCounts.print()
- }
- }
Flink中的DataStream程序是实现数据流转换的常规程序（例如 filtering, updating state, defining windows, aggregating）。最初从各种源（例如， message queues, socket streams, files）创建数据流。结果通过接收器返回，接收器可以例如将数据写入文件或标准输出（例如命令行终端）。Flink程序可以在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在很多计算机的集群上执行。
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}
object StreamDemo {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val file = env.socketTextStream("localhost", 9999)
val spliFile: DataStream[String] = file.flatMap(_.split(" "))
val wordAndOne: DataStream[(String, Int)] = spliFile.map((_, 1))
val keyed = wordAndOne.keyBy(data=>data._1)
val wordAndCount: DataStream[(String, Int)] = keyed.sum(1)
wordAndCount.print()
env.execute()
}
}
要运行示例程序，首先从终端使用netcat启动输入流：
nc -lk 9999
只需键入一些单词就可以返回一个新单词。这些将是字数统计程序的输入。
Flink程序提交到集群
1)Web提交方式：
图3 web提交方式
1)脚本方式：
- #!/bin/sh
- FLINK_HOME=/home/bigdata/apps/flink
- $FLINK_HOME/bin/flink run \
- -c BatchDemo \
- /root/wc.jar \
- hdfs://hadoop101:8020/wordcount/words.txt \
- hdfs://hadoop101:8020/wordcount/output3
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
