flink项目开发-flink的scala shell命令行交互模式开发

青旅半醒 2022-04-11 10:21 672阅读 0赞

# flink的 scala shell命令行交互模式开发 #

版本

Flink：1.6.2

flink带有一个集成的scala shell命令行。它可以以本地方式启动来模拟集群集群。执行下面的命令就可以通过shell命令行和flink集群交互(这种方式方便于代码调试)：

bin/start-scala-shell.sh local

如果想在集群上面运行scala shell，请查看本节后面的内容。

flink scala shell 用法  
shell方式支持流处理和批处理。当启动shell命令行之后，两个不同的ExecutionEnvironments会被自动创建。使用benv和senv分别去处理批处理和流处理程序。(类似于spark-shell中sc变量)

DataSet API  
下面的例子将会在scala shell中执行wordcount程序

输入benv按制表按键可以查看benv有哪些方法  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MDUwOTkz_size_16_color_FFFFFF_t_70][]

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,")
    Scala> val counts = text
        .flatMap { _.toLowerCase.split("\\W+") }
        .map { (_, 1) }.groupBy(0).sum(1)
    Scala-Flink> counts.print()

print()命令会自定发送指定的任务到jobmanager去执行，并且会将结果显示在控制台。

也可以吧结果写到一个文件中，然而，在这种情况下，你需要执行execute方法，去执行你的程序

Scala> benv.execute("MyProgram")

### DataStream API ###

类似于上面的批处理程序，我们可以通过DataStream API执行一个流处理程序。

同理：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MDUwOTkz_size_16_color_FFFFFF_t_70 1][]

Scala> 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,")
    Scala> val countsStreaming = textStreaming
        .flatMap { _.toLowerCase.split("\\W+") }
        .map { (_, 1) }.keyBy(0).sum(1)
    Scala> countsStreaming.print()
    Scala> senv.execute("Streaming Wordcount")

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MDUwOTkz_size_16_color_FFFFFF_t_70 2][]

注意：在流处理情况下，print方法不会触发执行。需要调用execute方法才会真正执行。

flink shell会自动带有命令执行历史。

在shell命令行模式下添加外部依赖  
可以将外部类路径添加到scala-shell中，当程序被调用的时候，这些外部依赖会自动的被发送到jobmanager上。

使用这个参数 -a <path/to/jar.jar> 或者 --addclasspath <path/to/jar.jar> 添加额外的依赖。

bin/start-scala-shell.sh [local | remote <host> <port> | yarn] --addclasspath <path/to/jar.jar>

## flink scala shell设置 ##

查看scala shell模式提供的选型，可以执行这个命令：

bin/start-scala-shell.sh --help

### 本地模式local ###

使用shell连接一个本地集成的flink 集群 使用下面命令

bin/start-scala-shell.sh local

### 远程模式remote ###

使用scala shell 连接一个远程集群，使用host和port参数去连接指定的jobmanager

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

集群模式 yarn scala shell cluster  
将hdfs文件注册成临时文件，然后再读取：

scala> benv.registerCachedFile("hdfs:///tmp/zjl/test1","hdfsFile")

scala> benv.re
    readCsvFile   readFileOfPrimitives   readTextFileWithValue   registerType                     
    readFile      readTextFile           registerCachedFile      registerTypeWithKryoSerializer

可以发现没有json读取方式，flink\_1.6.2读取格式有限。

scala> benv.readTextFile("hdfsFile")
    res1: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@739500b8

读取刚注册的连文件，返回的是DataSet格式。

scala> val text=benv.readTextFile("hdfsFile")
    text: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@14d7a962
    
    scala> text.
    aggregate       first                     join            minResources         rebalance            withForwardedFields         
    coGroup         flatMap                   joinWithHuge    name                 reduce               withForwardedFieldsFirst    
    collect         fullOuterJoin             joinWithTiny    output               reduceGroup          withForwardedFieldsSecond   
    combineGroup    getExecutionEnvironment   leftOuterJoin   partitionByHash      registerAggregator   withParameters              
    count           getParallelism            map             partitionByRange     rightOuterJoin       write                       
    cross           getType                   mapPartition    partitionCustom      setParallelism       writeAsCsv                  
    crossWithHuge   groupBy                   max             preferredResources   sortPartition        writeAsText                 
    crossWithTiny   iterate                   maxBy           print                sum                                              
    distinct        iterateDelta              min             printOnTaskManager   union                                            
    filter          iterateWithTermination    minBy           printToErr           withBroadcastSet

可以通过scala shell在yarn上启动一个专有的flink集群，yarn containers的数量可以通过参数-n 指定。shell在yarn上部署了一个新的集群并且连接到这个集群。你也可以指定集群的参数，例如：指定jobmanager的内存，yarn application的名称 等等。

例如：针对scala shell启动一个yarn集群包含两个taskmanager，使用下面的参数：

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

针对所有的参数选项，可以在本节的最后查看完整的说明

yarn session模式  
如果你之前已经使用flink yarn session模式启动了一个flink集群，scala shell可以使用下面的命令进行连接：

bin/start-scala-shell.sh yarn

完整的参数选项

Flink Scala Shell
    用法: start-scala-shell.sh [local|remote|yarn] [options] <args>...
     
    命令: local [options]
    使用scala shell连接一个本地flink集群
      -a <path/to/jar> | --addclasspath <path/to/jar>
            指定flink使用的第三方依赖
    命令: remote [options] <host> <port>
    启动scala shell连接一个远程集群
      <host>
            主机名
      <port>
            端口号
     
      -a <path/to/jar> | --addclasspath <path/to/jar>
            指定flink使用的第三方依赖
    命令: yarn [options]
    使用flink连接一个yarn集群
      -n arg | --container arg
            分配的yarn container的数量 (等于TaskManagers的数量)
      -jm arg | --jobManagerMemory arg
            JobManager container 的内存[in MB]
      -nm <value> | --name <value>
            在YARN上给应用设置一个名字
      -qu <arg> | --queue <arg>
            指定YARN队列
      -s <arg> | --slots <arg>
            指定每个TaskManager的slot数量
      -tm <arg> | --taskManagerMemory <arg>
            TaskManager container的内存 [in MB]
      -a <path/to/jar> | --addclasspath <path/to/jar>
            指定flink使用的第三方jar
      --configDir <value>
            配置文件目录.
      -h | --help
            打印帮助信息

更多内容请参考Flink官网：[https://flink.apache.org/][https_flink.apache.org]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MDUwOTkz_size_16_color_FFFFFF_t_70]: /images/20220411/c9e5cf2d3b0e4ac58c1314f5a80da24e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MDUwOTkz_size_16_color_FFFFFF_t_70 1]: /images/20220411/2c8b45caffe84238a3e2c61ab526fefb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MDUwOTkz_size_16_color_FFFFFF_t_70 2]: /images/20220411/a67a283532314d739c3fb29c770a83aa.png
[https_flink.apache.org]: https://flink.apache.org/