2. 环境准备
(2019-11-12)
2.1 项目模板
Flink执行环境分为本地环境和集群环境,一般地,为了正确执行flink需要JDK环境/Scala环境,Maven环境,Hadoop环境。
为了快速搭建,针对java和scala官网提供了项目模板来快速创建项目,官网参考地址:
java版本项目模板
scala版本项目模板
官网提供2种方式快速构建项目,maven和脚本方式(需要自行制定版本):
$ mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-scala \
-DarchetypeVersion=1.9.0
$ curl https://flink.apache.org/q/quickstart-scala.sh | bash -s 1.9.0
2.2 Flink开发环境
-
scala插件安装:
官方建议使用IDEA,所以需要安装。另外,需要安装Scala插件,具体方法如下:
Preferences -> Plugins
如果未安装过scala插件, 在MarketPlace 搜索scala,Install即可。
安装后重启IDE,完成插件的安装。
2.导入项目
项目创建后,使用IDEA打开,可以看到项目结构:
Java的默认JVM堆大小对于Flink可能太小,建议手工增加。
2.3 配置依赖
关于Maven依赖的scope设置的备用知识:maven依赖的scope含义
依赖配置项:
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<!-- Scala Library, provided by Flink as well. -->
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.version}</version>
<scope>provided</scope>
</dependency>
建议将依赖的scope项配置为“provided”,否则会使生成的JAR变得过大,因为它还包含所有Flink核心依赖项。更坏可能是,添加到应用程序jar文件中的Flink核心依赖项与您自己的某些依赖项版本发生冲突。
大多数flink应用程序都需要特定的连接器或库来运行,例如与Kafka,Cassandra等的连接器。添加如下依赖,并且打包时,一同打包上传。
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka-0.10_2.11</artifactId>
<version> 1.9.0 </version>
</dependency>
2.4 Scala REPL
官网参考:Scala REPL
为了实验此节,需要先安装一个本地,下载scala的安装包,直接解压:
$ tar zxvf flink-1.9.1-bin-scala_2.11.tgz
$ cd flink-1.9.1/
$ export TERM=xterm-color
$ bin/start-scala-shell.sh local #启动scala REPL 本地
Flink自带一个集成的scala shell交互,即Scala REPL。Scala REPL支持DataSet,DataStream,Table API和SQL。
上边启动Scala REPL遇到了几个坑:1.先不用本地启动集群,直接启动Scala REPL,否则的话报错。 2. 直接启动时,报了一个exception:java.lang.NumberFormatException: For input string: "0x100",想解决需要执行下$ export TERM=xterm-color,(由于终端一些样式格式的问题导致)。
当启动Scala REPL时,Flink已经初始化好了相应的Environment,分别使用“benv”和“senv”变量获取批量和流式计算环境。
下面是官网的例子来实现批量和流式的WordCount:
scala> val text = benv.fromElements(
| "To be, or not to be,--that is the question:--",
| "Whether 'tis nobler in the mind to suffer",
| "The slings and arrows of outrageous fortune",
| "Or to take arms against a sea of troubles,")
text: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@1f3f0d25
scala> val counts = text.flatMap{ _.toLowerCase.split("\\W+") }.map { (_, 1) }.groupBy(0).sum(1)
counts: org.apache.flink.api.scala.AggregateDataSet[(String, Int)] = org.apache.flink.api.scala.AggregateDataSet@83b0d9f
scala> counts.print()
1.在scala REPL交互式窗口,想退出的话,输入命令 :q
2.5 Flink源码编译
下载源码包,既可以从github上clone下来,也可以在官网download页直接下载源代码,源代码包很小。
$ git clone https://github.com/apache/flink
3. Flink编程模型
(2019-11-13)
有界数据集具有时间边界,无界数据集没有时间边界;对有界数据的处理称为批计算,对无界数据的处理称为流计算;支持批处理的API称为DataSet API,支持流处理的API称为DataStream API。
网友评论