Flink Scala Shell：使用交互式编程环境学习和调试Flink 三种模式 local remote yarn

比眉伴天荒 2022-11-06 05:51 12阅读 0赞

## 交互式编程环境：REPL ##

当前最著名的交互式编程环境莫属Jupyter Notebook了，程序员可以启动一个交互的Session，在这Session中编写代码、执行程序、获取结果，所见即所得。

交互式编程的优势包括：

*  快速启动一个程序：程序员不需要“编译、打包、执行”这一套复杂过程，只需要开启一个交互Session，敲下代码，直接得到结果，非常适合验证一段代码的结果。
 *  直接获得程序反馈：使用`print`，可以在交互环境中直接得到程序结果，无需将输出导出到文件或其他位置。

程序员敲入几行程序命令，环境可以立刻给出反馈，因此这种交互式环境非常适合调试程序，尤其对于初学者来说非常友好。交互式编程环境通常被称为REPL（Read-Eval-Print Loop），这种方式跟Linux的命令行非常相似，因此又被称为Shell。支持REPL的语言有Python、R、Matlab、Scala以及Java 9。

我之前经常使用Spark的交互式环境spark-shell，Flink基于Java和Scala，其实也是支持交互式编程的，这里推荐新人使用REPL交互式环境来上手和学习Flink。

**注意，Flink的交互式编程环境只支持Scala语言，程序员可以基于Scala语言调用DataStream/DataSet API、Table API & SQL，不支持Java。另外，Flink提供了Python版本的REPL环境，不过目前Flink（1.9）的Python API只支持Table API调用。**本文主要展示Scala的REPL的使用方法。

## 启动REPL环境 ##

### 下载Flink ###

前往Flink Downloads页面（[https://flink.apache.org/downloads.html][https_flink.apache.org_downloads.html]）下载编译好的Flink程序，这里根据你的Scala版本、是否需要搭载Hadoop环境等需求来选择适合的版本，没有特殊需求的选择最近版本的Flink即可。下载的tgz包解压缩后，在`bin`目录下有一些Flink提供的基础工具。

注意，Flink目前对类Unix系统（Linux、macOS）比较友好，尽管Flink可以在Windows默认命令行上运行，但只限于一些基础命令，比如不支持REPL。Windows用户需要使用Cygwin、Windows Subsystem for Linux等方式来模拟Unix环境。

### 启动REPL ###

在命令行里进入Flink解压缩之后的目录，在本地启动一个Flink REPL交互式环境。

bin/start-scala-shell.sh local

启动后，命令行中会反馈一些注意信息：

![e5d3c6a361484e9b35be6cc6488322fe.png][]

验证一下Scala的Hello World：

scala> print("Hello World!")
    Hello World!

## Scala Shell的使用 ##

### 使用正确的运行环境 ###

Flink Shell已经支持批处理和流处理两种模式。如上图所示，Flink在这个交互环境中默认提供运行环境Execution Environment，其中批处理为`benv`、流处理为`senv`。Flink是一个运行在集群上的大数据系统，需要通过运行环境这个入口与集群交互，因此运行环境是Flink程序必不可少的重要元素。

下面演示使用Scala Shell来运行一个最基础的map算子：

scala> val dataStream: DataStream[Int] = senv.fromElements(1, 2, -3, 0, 5, -9, 8)
    dataStream: org.apache.flink.streaming.api.scala.DataStream[Int] = org.apache.flink.streaming.api.scala.DataStream@3013e1e8
    
    scala> val lambda = dataStream.map ( input => input * 2 ).print()
    lambda: org.apache.flink.streaming.api.datastream.DataStreamSink[Int] = org.apache.flink.streaming.api.datastream.DataStreamSink@4ff00844
    
    scala> senv.execute("Basic Map Transformation")
    2
    4
    -6
    0
    10
    -18
    16
    res0: org.apache.flink.api.common.JobExecutionResult = org.apache.flink.api.common.JobExecutionResult@7f59f4e4

我创建了一个数字列表DataStream，然后使用map对每个元素乘以2，并打印出来。

==注意，在流处理模式下，print不会自动触发，必须调用`execute`才能触发执行前面的程序。==

### 代码拷贝 ###

我们经常遇到的一个使用场景是从网上看到一些代码片段，需要拷贝过来验证正确性。在Scala Shell中，可以使用`:paste`命令进入拷贝模式，复制粘贴之后，再使用`Control + D`按键组合退出粘贴模式。

scala> :paste
    // Entering paste mode (ctrl-D to finish)
    
    val textStreaming = senv.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,")
    ^D
    // Exiting paste mode, now interpreting.
    
    textStreaming: org.apache.flink.streaming.api.scala.DataStream[String] = org.apache.flink.streaming.api.scala.DataStream@62e8ef9f

### 使用其他依赖 ###

如果程序依赖了其他包，可以在启动Flink Scala Shell时，加上参数`-a <path/to/jar>`或`--addclasspath <path/to/jar>`。

例如，我想使用Gson来解析json数据：

bin/start-scala-shell.sh local -a /Users/luweizheng/.m2/repository/com/google/code/gson/gson/2.8.5/gson-2.8.5.jar

这样我就能在交互式环境中使用这个包下的各种类和方法了。

绝大多数情况下，我们可能要依赖多个不同的包，这时候需要使用`maven-shade-plugin`工具将所依赖包合并到一起，打成一个超级包（uber-jar），超级包内包含了这个程序所有必备的依赖。

### 使用Flink集群 ###

Flink Scala Shell也支持集群模式，包括独立的Flink集群和与其他应用共享的Yarn集群。

### 远程Flink集群 ###

使用remote模式，指定JobManager的机器名（IP）和端口号：

bin/start-scala-shell.sh remote <hostname> <portnumber>

### Yarn集群 ###

使用这个命令可以在Yarn上部署一个新的Flink集群，并使用其他参数来配置集群信息，比如\`-n 2将申请2个TaskManager，其他详细使用方法可以参见下面完整使用手册。

bin/start-scala-shell.sh yarn -n 2

## 完整使用方法 ##

Flink Scala Shell
    Usage: start-scala-shell.sh [local|remote|yarn] [options] <args>...
    
    Command: local [options]
    Starts Flink scala shell with a local Flink cluster
      -a <path/to/jar> | --addclasspath <path/to/jar>
            Specifies additional jars to be used in Flink
    Command: remote [options] <host> <port>
    Starts Flink scala shell connecting to a remote cluster
      <host>
            Remote host name as string
      <port>
            Remote port as integer
    
      -a <path/to/jar> | --addclasspath <path/to/jar>
            Specifies additional jars to be used in Flink
    Command: yarn [options]
    Starts Flink scala shell connecting to a yarn cluster
      -n arg | --container arg
            Number of YARN container to allocate (= Number of TaskManagers)
      -jm arg | --jobManagerMemory arg
            Memory for JobManager container with optional unit (default: MB)
      -nm <value> | --name <value>
            Set a custom name for the application on YARN
      -qu <arg> | --queue <arg>
            Specifies YARN queue
      -s <arg> | --slots <arg>
            Number of slots per TaskManager
      -tm <arg> | --taskManagerMemory <arg>
            Memory per TaskManager container with optional unit (default: MB)
      -a <path/to/jar> | --addclasspath <path/to/jar>
            Specifies additional jars to be used in Flink
      --configDir <value>
            The configuration directory.
      -h | --help
            Prints this usage text

[https_flink.apache.org_downloads.html]: https://link.zhihu.com/?target=https%3A//flink.apache.org/downloads.html
[e5d3c6a361484e9b35be6cc6488322fe.png]: /images/20221023/57652e86b1d34d4ba2be539b222b14cc.png