learning flink

2019-11-14 本文已影响0人奉先

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。