devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

5.Flink之客户端操作_org.apache.flink.client.program.rest.restclustercl

作者：你好赵伟 | 2024-05-07 10:34:41

赞

踩

org.apache.flink.client.program.rest.restclusterclient

1. 环境说明

在前面几期的课程里面讲过了 Flink 开发环境的搭建和应用的部署以及运行，今天的课程主要是讲 Flink 的客户端操作。本次讲解以实际操作为主。这次课程是基于社区的 Flink 1.7.2 版本，操作系统是 Mac 系统，浏览器是 Google Chrome 浏览器。有关开发环境的准备和集群的部署，请参考「开发环境搭建和应用的配置、部署及运行」的内容。

2. 课程概要

如下图所示，Flink 提供了丰富的客户端操作来提交任务和与任务进行交互，包括 Flink 命令行，Scala Shell，SQL Client，Restful API 和 Web。Flink 首先提供的最重要的是命令行，其次是 SQL Client 用于提交 SQL 任务的运行，还有就是 Scala Shell 提交 Table API 的任务。同时，Flink 也提供了Restful 服务，用户可以通过 http 方式进行调用。此外，还有 Web 的方式可以提交任务。

在 Flink 安装目录的 bin 目录下面可以看到有 flink, start-scala-shell.sh 和 sql-client.sh 等文件，这些都是客户端操作的入口。

3.1 Flink 命令行

Flink 的命令行参数很多，输入 flink – h 能看到完整的说明：

➜  flink-1.7.2 bin/flink -h

如果想看某一个命令的参数，比如 Run 命令，输入：

➜  flink-1.7.2 bin/flink run -h

本文主要讲解常见的一些操作，更详细的文档请参考: Flink 命令行官方文档。

3.1.1 Standalone

首先启动一个 Standalone 的集群：

  1. ➜  flink-1.7.2 bin/start-cluster.sh
  2. Starting cluster.
  3. Starting standalonesession daemon on host zkb-MBP.local.
  4. Starting taskexecutor daemon on host zkb-MBP.local.

打开 http://127.0.0.1:8081 能看到 Web 界面。

Run

运行任务，以 Flink 自带的例子 TopSpeedWindowing 为例：

  1. ➜  flink-1.7.2 bin/flink run -d examples/streaming/TopSpeedWindowing.jar
  2. Starting execution of program
  3. Executing TopSpeedWindowing example with default input data set.
  4. Use --input to specify file input.
  5. Printing result to stdout. Use --output to specify output path.
  6. Job has been submitted with JobID 5e20cb6b0f357591171dfcca2eea09de

运行起来后默认是 1 个并发:

点左侧「Task Manager」，然后点「Stdout」能看到输出日志：

或者查看本地 Log 目录下的 *.out 文件：

List

查看任务列表：

  1. ➜  flink-1.7.2 bin/flink list -m 127.0.0.1:8081
  2. Waiting for response...
  3. ------------------ Running/Restarting Jobs -------------------
  4. 24.03.2019 10:14:06 : 5e20cb6b0f357591171dfcca2eea09de : CarTopSpeedWindowingExample (RUNNING)
  5. --------------------------------------------------------------
  6. No scheduled jobs.

Stop

停止任务。通过 -m 来指定要停止的 JobManager 的主机地址和端口。

  1. ➜  flink-1.7.2 bin/flink stop -m 127.0.0.1:8081 d67420e52bd051fae2fddbaa79e046bb
  2. Stopping job d67420e52bd051fae2fddbaa79e046bb.
  3. ------------------------------------------------------------
  4. The program finished with the following exception:
  5.   org.apache.flink.util.FlinkException: Could not stop the job   d67420e52bd051fae2fddbaa79e046bb.
  6.   at org.apache.flink.client.cli.CliFrontend.lambda$stop$5(CliFrontend.java:554)
  7.   at org.apache.flink.client.cli.CliFrontend.runClusterAction(CliFrontend.java:985)
  8.   at org.apache.flink.client.cli.CliFrontend.stop(CliFrontend.java:547)
  9.   at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1062)
  10.   at org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1126)
  11.   at java.security.AccessController.doPrivileged(Native Method)
  12.   at javax.security.auth.Subject.doAs(Subject.java:422)
  13.   at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
  14.   at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
  15.   at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1126)
  16. Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.rest.util.RestClientException: [Job termination (STOP) failed: This job is not stoppable.]
  17.   at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
  18.   at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915)
  19.   at org.apache.flink.client.program.rest.RestClusterClient.stop(RestClusterClient.java:392)
  20.   at org.apache.flink.client.cli.CliFrontend.lambda$stop$5(CliFrontend.java:552)
  21. ... 9 more
  22. Caused by: org.apache.flink.runtime.rest.util.RestClientException: [Job termination (STOP) failed: This job is not stoppable.]
  23.   at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:380)
  24.   at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:364)
  25.   at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952)
  26.   at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926)
  27.   at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)
  28.   at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  29.   at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  30.   at java.lang.Thread.run(Thread.java:748)

从日志里面能看出 Stop 命令执行失败了。一个 Job 能够被 Stop 要求所有的 Source 都是可以 Stoppable 的，即实现了 StoppableFunction 接口。

  1. /**
  2.  * 需要能 stoppable 的函数必须实现这个接口，例如流式任务的 source。
  3.  * stop() 方法在任务收到 STOP 信号的时候调用。
  4.  * source 在接收到这个信号后，必须停止发送新的数据且优雅的停止。
  5.  */
  6. @PublicEvolving
  7. public interface StoppableFunction {
  8.     /**
  9.       * 停止 source。与 cancel() 不同的是，这是一个让 source 优雅停止的请求。
  10.      * 等待中的数据可以继续发送出去，不需要立即停止。
  11.       */
  12.     void stop();
  13. }

Cancel

取消任务。如果在 conf/flink-conf.yaml 里面配置了 state.savepoints.dir，会保存 Savepoint，否则不会保存 Savepoint。

  1. ➜  flink-1.7.2 bin/flink cancel -m 127.0.0.1:8081 5e20cb6b0f357591171dfcca2eea09de
  2.  
  3. Cancelling job 5e20cb6b0f357591171dfcca2eea09de.
  4. Cancelled job 5e20cb6b0f357591171dfcca2eea09de.

也可以在停止的时候显示指定 Savepoint 目录。

  1. ➜  flink-1.7.2 bin/flink cancel -m 127.0.0.1:8081 -s /tmp/savepoint 29da945b99dea6547c3fbafd57ed8759
  2.  
  3. Cancelling job 29da945b99dea6547c3fbafd57ed8759 with savepoint to /tmp/savepoint.
  4. Cancelled job 29da945b99dea6547c3fbafd57ed8759. Savepoint stored in file:/tmp/savepoint/savepoint-29da94-88299bacafb7.
  5.  
  6. ➜  flink-1.7.2 ll /tmp/savepoint/savepoint-29da94-88299bacafb7
  7. total 32K
  8. -rw-r--r-- 1 baoniu 29K Mar 24 10:33 _metadata

取消和停止（流作业）的区别如下：

cancel() 调用，立即调用作业算子的 cancel() 方法，以尽快取消它们。如果算子在接到 cancel() 调用后没有停止，Flink 将开始定期中断算子线程的执行，直到所有算子停止为止。

stop() 调用，是更优雅的停止正在运行流作业的方式。stop() 仅适用于 Source 实现了 StoppableFunction 接口的作业。当用户请求停止作业时，作业的所有 Source 都将接收 stop() 方法调用。直到所有 Source 正常关闭时，作业才会正常结束。这种方式，使作业正常处理完所有作业。

Savepoint

触发 Savepoint。

  1. ➜  flink-1.7.2 bin/flink savepoint -m 127.0.0.1:8081 ec53edcfaeb96b2a5dadbfbe5ff62bbb /tmp/savepoint
  2. Triggering savepoint for job ec53edcfaeb96b2a5dadbfbe5ff62bbb.
  3. Waiting for response...
  4. Savepoint completed. Path: file:/tmp/savepoint/savepoint-ec53ed-84b00ce500ee
  5. You can resume your program from this savepoint with the run command.

说明：Savepoint 和 Checkpoint 的区别（详见文档）：

Checkpoint 是增量做的，每次的时间较短，数据量较小，只要在程序里面启用后会自动触发，用户无须感知；Checkpoint 是作业 failover 的时候自动使用，不需要用户指定。

Savepoint 是全量做的，每次的时间较长，数据量较大，需要用户主动去触发。Savepoint 一般用于程序的版本更新（详见文档），Bug 修复，A/B Test 等场景，需要用户指定。

通过 -s 参数从指定的 Savepoint 启动：

  1. ➜  flink-1.7.2 bin/flink run -d -s /tmp/savepoint/savepoint-f049ff-24ec0d3e0dc7 ./examples/streaming/TopSpeedWindowing.jar
  2. Starting execution of program
  3. Executing TopSpeedWindowing example with default input data set.
  4. Use --input to specify file input.
  5. Printing result to stdout. Use --output to specify output path.

查看 JobManager 的日志，能够看到类似这样的 Log：

  1. 2019-03-28 10:30:53,957 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     
  2. - Starting job 790d7b98db6f6af55d04aec1d773852d from savepoint /tmp/savepoint/savepoint-f049ff-24ec0d3e0dc7 ()
  3. 2019-03-28 10:30:53,959 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator    
  4.  - Reset the checkpoint ID of job 790d7b98db6f6af55d04aec1d773852d to 2.
  5. 2019-03-28 10:30:53,959 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     
  6. - Restoring job 790d7b98db6f6af55d04aec1d773852d from latest valid checkpoint: Checkpoint 1 @ 0 for 790d7b98db6f6af55d04aec1d773852d.

Modify

修改任务并行度。

为了方便演示，我们修改 conf/flink-conf.yaml 将 Task Slot 数从默认的 1 改为 4，并配置 Savepoint 目录。（Modify 参数后面接 -s 指定 Savepoint 路径当前版本可能有 Bug，提示无法识别）

  1. taskmanager.numberOfTaskSlots: 4
  2. state.savepoints.dir: file:///tmp/savepoint

修改参数后需要重启集群生效，然后再启动任务：

  1. ➜  flink-1.7.2 bin/stop-cluster.sh && bin/start-cluster.sh
  2. Stopping taskexecutor daemon (pid: 53139) on host zkb-MBP.local.
  3. Stopping standalonesession daemon (pid: 52723) on host zkb-MBP.local.
  4. Starting cluster.
  5. Starting standalonesession daemon on host zkb-MBP.local.
  6. Starting taskexecutor daemon on host zkb-MBP.local.
  7.  
  8. ➜  flink-1.7.2 bin/flink run -d examples/streaming/TopSpeedWindowing.jar
  9. Starting execution of program
  10. Executing TopSpeedWindowing example with default input data set.
  11. Use --input to specify file input.
  12. Printing result to stdout. Use --output to specify output path.
  13. Job has been submitted with JobID 7752ea7b0e7303c780de9d86a5ded3fa

从页面上能看到 Task Slot 变为了 4，这时候任务的默认并发度是 1。

通过 Modify 命令依次将并发度修改为 4 和 3，可以看到每次 Modify 命令都会触发一次 Savepoint。

  1. ➜  flink-1.7.2 bin/flink modify -p 4 7752ea7b0e7303c780de9d86a5ded3fa
  2. Modify job 7752ea7b0e7303c780de9d86a5ded3fa.
  3. Rescaled job 7752ea7b0e7303c780de9d86a5ded3fa. Its new parallelism is 4.
  4.  
  5. ➜  flink-1.7.2 ll /tmp/savepoint
  6. total 0
  7. drwxr-xr-x 3 baoniu 96 Jun 17 09:05 savepoint-7752ea-00c05b015836/
  8.  
  9. ➜  flink-1.7.2 bin/flink modify -p 3 7752ea7b0e7303c780de9d86a5ded3fa
  10. Modify job 7752ea7b0e7303c780de9d86a5ded3fa.
  11. Rescaled job 7752ea7b0e7303c780de9d86a5ded3fa. Its new parallelism is 3.
  12.  
  13. ➜  flink-1.7.2 ll /tmp/savepoint
  14. total 0
  15. drwxr-xr-x 3 baoniu 96 Jun 17 09:08 savepoint-7752ea-449b131b2bd4/

查看 JobManager 的日志，可以看到：

  1. 2019-06-17 09:05:11,179 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Starting job 7752ea7b0e7303c780de9d86a5ded3fa from savepoint file:/tmp/savepoint/savepoint-790d7b-3581698f007e ()
  2. 2019-06-17 09:05:11,182 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Reset the checkpoint ID of job 7752ea7b0e7303c780de9d86a5ded3fa to 3.
  3. 2019-06-17 09:05:11,182 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - Restoring job 790d7b98db6f6af55d04aec1d773852d from latest valid checkpoint: Checkpoint 2 @ 0 for 7752ea7b0e7303c780de9d86a5ded3fa.
  4. 2019-06-17 09:05:11,184 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator     - No master state to restore
  5. 2019-06-17 09:05:11,184 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph        - Job CarTopSpeedWindowingExample (7752ea7b0e7303c780de9d86a5ded3fa) switched from state RUNNING to SUSPENDING.
  6. org.apache.flink.util.FlinkException: Job is being rescaled.

Info

Info 命令是用来查看 Flink 任务的执行计划（StreamGraph）的。

  1. ➜  flink-1.7.2 bin/flink info examples/streaming/TopSpeedWindowing.jar
  2. ----------------------- Execution Plan -----------------------
  3. {"nodes":[{"id":1,"type":"Source: Custom Source","pact":"Data Source","contents":"Source: Custom Source","parallelism":1},{"id":2,"type":"Timestamps/Watermarks","pact":"Operator","contents":"Timestamps/Watermarks","parallelism":1,"predecessors":[{"id":1,"ship_strategy":"FORWARD","side":"second"}]},{"id":4,"type":"Window(GlobalWindows(), DeltaTrigger, TimeEvictor, ComparableAggregator, PassThroughWindowFunction)","pact":"Operator","contents":"Window(GlobalWindows(), DeltaTrigger, TimeEvictor, ComparableAggregator, PassThroughWindowFunction)","parallelism":1,"predecessors":[{"id":2,"ship_strategy":"HASH","side":"second"}]},{"id":5,"type":"Sink: Print to Std. Out","pact":"Data Sink","contents":"Sink: Print to Std. Out","parallelism":1,"predecessors":[{"id":4,"ship_strategy":"FORWARD","side":"second"}]}]}
  4. --------------------------------------------------------------

拷贝输出的 Json 内容，粘贴到这个网站：http://flink.apache.org/visualizer/

可以和实际运行的物理执行计划对比：

3.1.2 Yarn per-job

单任务 Attach 模式

默认是 Attach 模式，即客户端会一直等待直到程序结束才会退出。

  • 通过 -m yarn-cluster 指定 Yarn 模式

  • Yarn 上显示名字为 Flink session cluster，这个 Batch 的 Wordcount 任务运行完会 FINISHED。

  • 客户端能看到结果输出

  1. [admin@z17.sqa.zth /home/admin/flink/flink-1.7.2]
  2. $echo $HADOOP_CONF_DIR
  3. /etc/hadoop/conf/
  4.  
  5. [admin@z17.sqa.zth /home/admin/flink/flink-1.7.2]
  6. $./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar
  7.  
  8. 2019-06-17 09:15:24,511 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at z05c05217.sqa.zth.tbsite.net/11.163.188.29:8050
  9. 2019-06-17 09:15:24,690 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  10. 2019-06-17 09:15:24,690 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  11. 2019-06-17 09:15:24,907 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=4}
  12. 2019-06-17 09:15:25,430 WARN  org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory       - The short-circuit local reads feature cannot be used because libhadoop cannot be loaded.
  13. 2019-06-17 09:15:25,438 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
  14. 2019-06-17 09:15:36,239 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1532332183347_0724
  15. 2019-06-17 09:15:36,276 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1532332183347_0724
  16. 2019-06-17 09:15:36,276 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
  17. 2019-06-17 09:15:36,281 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
  18. 2019-06-17 09:15:40,426 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
  19. Starting execution of program
  20. Executing WordCount example with default input data set.
  21. Use --input to specify file input.
  22. Printing result to stdout. Use --output to specify output path.
  23. (a,5)
  24. (action,1)
  25. (after,1)
  26. (against,1)
  27. (all,2)
  28. ... ...
  29. (would,2)
  30. (wrong,1)
  31. (you,1)
  32. Program execution finished
  33. Job with JobID 8bfe7568cb5c3254af30cbbd9cd5971e has finished.
  34. Job Runtime: 9371 ms
  35. Accumulator Results:
  36. - 2bed2c5506e9237fb85625416a1bc508 (java.util.ArrayList) [170 elements]

如果我们以 Attach 模式运行 Streaming 的任务，客户端会一直等待不退出，可以运行以下的例子试验下：

./bin/flink run -m yarn-cluster ./examples/streaming/TopSpeedWindowing.jar

单任务 Detached 模式

  • 由于是 Detached 模式，客户端提交完任务就退出了

  • Yarn 上显示为 Flink per-job cluster

  1. $./bin/flink run -yd -m yarn-cluster ./examples/streaming/TopSpeedWindowing.jar
  2.  
  3. 2019-06-18 09:21:59,247 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at z05c05217.sqa.zth.tbsite.net/11.163.188.29:8050
  4. 2019-06-18 09:21:59,428 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  5. 2019-06-18 09:21:59,428 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  6. 2019-06-18 09:21:59,940 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=4}
  7. 2019-06-18 09:22:00,427 WARN  org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory       - The short-circuit local reads feature cannot be used because libhadoop cannot be loaded.
  8. 2019-06-18 09:22:00,436 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
  9. ^@2019-06-18 09:22:12,113 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1532332183347_0729
  10. 2019-06-18 09:22:12,151 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1532332183347_0729
  11. 2019-06-18 09:22:12,151 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
  12. 2019-06-18 09:22:12,155 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
  13. 2019-06-18 09:22:16,275 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
  14. 2019-06-18 09:22:16,275 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The Flink YARN client has been started in detached mode. In order to stop Flink on YARN, use the following command or a YARN web interface to stop it:
  15. yarn application -kill application_1532332183347_0729
  16. Please also note that the temporary files of the YARN session in the home directory will not be removed.
  17. Job has been submitted with JobID e61b9945c33c300906ad50a9a11f36df

3.1.3 Yarn session

启动 Session

./bin/yarn-session.sh -tm 2048 -s 3

表示启动一个 Yarn session 集群，每个 TM 的内存是 2 G，每个 TM 有 3 个 Slot。(注意：-n 参数不生效)

  1. ➜  flink-1.7.2 ./bin/yarn-session.sh -tm 2048 -s 3
  2. 2019-06-17 09:21:50,177 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, localhost
  3. 2019-06-17 09:21:50,179 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
  4. 2019-06-17 09:21:50,179 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
  5. 2019-06-17 09:21:50,179 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
  6. 2019-06-17 09:21:50,179 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 4
  7. 2019-06-17 09:21:50,179 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: state.savepoints.dir, file:///tmp/savepoint
  8. 2019-06-17 09:21:50,180 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
  9. 2019-06-17 09:21:50,180 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: rest.port, 8081
  10. 2019-06-17 09:21:50,644 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
  11. 2019-06-17 09:21:50,746 INFO  org.apache.flink.runtime.security.modules.HadoopModule        - Hadoop user set to baoniu (auth:SIMPLE)
  12. 2019-06-17 09:21:50,848 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at z05c05217.sqa.zth.tbsite.net/11.163.188.29:8050
  13. 2019-06-17 09:21:51,148 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=2048, numberTaskManagers=1, slotsPerTaskManager=3}
  14. 2019-06-17 09:21:51,588 WARN  org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory       - The short-circuit local reads feature cannot be used because libhadoop cannot be loaded.
  15. 2019-06-17 09:21:51,596 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
  16. ^@2019-06-17 09:22:03,304 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1532332183347_0726
  17. 2019-06-17 09:22:03,336 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1532332183347_0726
  18. 2019-06-17 09:22:03,336 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
  19. 2019-06-17 09:22:03,340 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
  20. 2019-06-17 09:22:07,722 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
  21. 2019-06-17 09:22:08,050 INFO  org.apache.flink.runtime.rest.RestClient                      - Rest client endpoint started.
  22. Flink JobManager is now running on z07.sqa.net:37109 with leader id 00000000-0000-0000-0000-000000000000.
  23. JobManager Web Interface: http://z07.sqa.net:37109

客户端默认是 Attach 模式，不会退出：

  • 可以 ctrl + c 退出，然后再通过 ./bin/yarn-session.sh -id application_1532332183347_0726 连上来；

  • 或者启动的时候用 -d 则为 detached 模式 

    Yarn 上显示为 Flink session cluster；

  • 在本机的临时目录（有些机器是 /tmp 目录）下会生成一个文件：

  1. ➜  flink-1.7.2 cat /var/folders/2b/r6d49pcs23z43b8fqsyz885c0000gn/T/.yarn-properties-baoniu
  2. #Generated YARN properties file
  3. #Mon Jun 17 09:22:08 CST 2019
  4. parallelism=3
  5. dynamicPropertiesString=
  6. applicationID=application_1532332183347_0726

提交任务

./bin/flink run ./examples/batch/WordCount.jar

将会根据 /tmp/.yarn-properties-admin 文件内容提交到了刚启动的 Session。

  1. ➜  flink-1.7.2 ./bin/flink run ./examples/batch/WordCount.jar
  2. 2019-06-17 09:26:42,767 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /var/folders/2b/r6d49pcs23z43b8fqsyz885c0000gn/T/.yarn-properties-baoniu.
  3. 2019-06-17 09:26:42,767 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /var/folders/2b/r6d49pcs23z43b8fqsyz885c0000gn/T/.yarn-properties-baoniu.
  4. 2019-06-17 09:26:43,058 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - YARN properties set default parallelism to 3
  5. 2019-06-17 09:26:43,058 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - YARN properties set default parallelism to 3
  6. YARN properties set default parallelism to 3
  7. 2019-06-17 09:26:43,097 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at z05c05217.sqa.zth.tbsite.net/11.163.188.29:8050
  8. 2019-06-17 09:26:43,229 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  9. 2019-06-17 09:26:43,229 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  10. 2019-06-17 09:26:43,327 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Found application JobManager host name 'z05c07216.sqa.zth.tbsite.net' and port '37109' from supplied application id 'application_1532332183347_0726'
  11. Starting execution of program
  12. Executing WordCount example with default input data set.
  13. Use --input to specify file input.
  14. Printing result to stdout. Use --output to specify output path.
  15. ^@(a,5)
  16. (action,1)
  17. (after,1)
  18. (against,1)
  19. (all,2)
  20. (and,12)
  21. ... ...
  22. (wrong,1)
  23. (you,1)
  24. Program execution finished
  25. Job with JobID ad9b0f1feed6d0bf6ba4e0f18b1e65ef has finished.
  26. Job Runtime: 9152 ms
  27. Accumulator Results:
  28. - fd07c75d503d0d9a99e4f27dd153114c (java.util.ArrayList) [170 elements]

运行结束后 TM 的资源会释放。

提交到指定的 Session

通过 -yid 参数来提交到指定的 Session。

  1. $./bin/flink run -d -p 30 -m yarn-cluster -yid application_1532332183347_0708 ./examples/streaming/TopSpeedWindowing.jar
  2.  
  3. 2019-03-24 12:36:33,668 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at z05c05217.sqa.zth.tbsite.net/11.163.188.29:8050
  4. 2019-03-24 12:36:33,773 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  5. 2019-03-24 12:36:33,773 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  6. 2019-03-24 12:36:33,837 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Found application JobManager host name 'z05c05218.sqa.zth.tbsite.net' and port '60783' from supplied application id 'application_1532332183347_0708'
  7. Starting execution of program
  8. Executing TopSpeedWindowing example with default input data set.
  9. Use --input to specify file input.
  10. Printing result to stdout. Use --output to specify output path.
  11. Job has been submitted with JobID 58d5049ebbf28d515159f2f88563f5fd

注：Blink版本 的 Session 与 Flink 的 Session 的区别：

  • Flink 的 session -n 参数不生效，而且不会提前启动 TM；
  • Blink 的 session 可以通过 -n 指定启动多少个 TM，而且 TM 会提前起来；

3.2 Scala Shell

官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.7/ops/scala_shell.html

3.2.1 Deploy

Local

  1. $bin/start-scala-shell.sh local
  2. Starting Flink Shell:
  3. Starting local Flink cluster (host: localhost, port: 8081).
  4. Connecting to Flink cluster (host: localhost, port: 8081).
  5. ... ...
  6. scala>

任务运行说明：

  • Batch 任务内置了 benv 变量，通过 print() 将结果输出到控制台；

  • Streaming 任务内置了 senv 变量，通过 senv.execute(“job name”) 来提交任务，且 Datastream 的输出只有在 Local 模式下打印到控制台；

Remote

先启动一个 yarn session cluster：

  1. $./bin/yarn-session.sh  -tm 2048 -s 3
  2.  
  3. 2019-03-25 09:52:16,341 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, localhost
  4. 2019-03-25 09:52:16,342 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
  5. 2019-03-25 09:52:16,342 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
  6. 2019-03-25 09:52:16,343 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
  7. 2019-03-25 09:52:16,343 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 4
  8. 2019-03-25 09:52:16,343 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
  9. 2019-03-25 09:52:16,343 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: state.savepoints.dir, file:///tmp/savepoint
  10. 2019-03-25 09:52:16,343 INFO  org.apache.flink.configuration.GlobalConfiguration            
  11. … ...
  12. Flink JobManager is now running on z054.sqa.net:28665 with leader id 00000000-0000-0000-0000-000000000000.
  13. JobManager Web Interface: http://z054.sqa.net:28665

启动 scala shell，连到 jm：

  1. $bin/start-scala-shell.sh remote z054.sqa.net 28665
  2. Starting Flink Shell:
  3. Connecting to Flink cluster (host: z054.sqa.net, port: 28665).
  4. ... ...
  5. scala>

Yarn

  1. $./bin/start-scala-shell.sh yarn -n 2 -jm 1024 -s 2 -tm 1024 -nm flink-yarn
  2.  
  3. Starting Flink Shell:
  4. 2019-03-25 09:47:44,695 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, localhost
  5. 2019-03-25 09:47:44,697 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
  6. 2019-03-25 09:47:44,697 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
  7. 2019-03-25 09:47:44,697 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
  8. 2019-03-25 09:47:44,697 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 4
  9. 2019-03-25 09:47:44,698 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
  10. 2019-03-25 09:47:44,698 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: state.savepoints.dir, file:///tmp/savepoint
  11. 2019-03-25 09:47:44,698 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: rest.port, 8081
  12. 2019-03-25 09:47:44,717 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-admin.
  13. 2019-03-25 09:47:45,041 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at z05c05217.sqa.zth.tbsite.net/11.163.188.29:8050
  14. 2019-03-25 09:47:45,098 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
  15. 2019-03-25 09:47:45,266 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
  16. 2019-03-25 09:47:45,275 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The argument yn is deprecated in will be ignored.
  17. 2019-03-25 09:47:45,357 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=2, slotsPerTaskManager=2}
  18. 2019-03-25 09:47:45,711 WARN  org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory       - The short-circuit local reads feature cannot be used because libhadoop cannot be loaded.
  19. 2019-03-25 09:47:45,718 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/admin/flink/flink-1.7.2/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
  20. 2019-03-25 09:47:46,514 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1532332183347_0710
  21. 2019-03-25 09:47:46,534 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1532332183347_0710
  22. 2019-03-25 09:47:46,534 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
  23. 2019-03-25 09:47:46,535 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
  24. 2019-03-25 09:47:51,051 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
  25. 2019-03-25 09:47:51,222 INFO  org.apache.flink.runtime.rest.RestClient                      - Rest client endpoint started.
  26.  
  27. Connecting to Flink cluster (host: 10.10.10.10, port: 56942).

按 CTRL + C 退出 Shell 后，这个 Flink cluster 还会继续运行，不会退出。

3.2.2 Execute

DataSet

  1. ➜  flink-1.7.2 bin/stop-cluster.sh
  2. No taskexecutor daemon to stop on host zkb-MBP.local.
  3. No standalonesession daemon to stop on host zkb-MBP.local.
  4. ➜  flink-1.7.2 bin/start-scala-shell.sh local
  5. Starting Flink Shell:
  6. Starting local Flink cluster (host: localhost, port: 8081).
  7. Connecting to Flink cluster (host: localhost, port: 8081).
  8.  
  9. scala> val text = benv.fromElements("To be, or not to be,--that is the question:--")
  10. text: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@5b407336
  11.  
  12. scala> val counts = text.flatMap { _.toLowerCase.split("\\W+") }.map { (_, 1) }.groupBy(0).sum(1)
  13. counts: org.apache.flink.api.scala.AggregateDataSet[(String, Int)] = org.apache.flink.api.scala.AggregateDataSet@6ee34fe4
  14.  
  15. scala> counts.print()
  16. (be,2)
  17. (is,1)
  18. (not,1)
  19. (or,1)
  20. (question,1)
  21. (that,1)
  22. (the,1)
  23. (to,2)

对 DataSet 任务来说，print() 会触发任务的执行。

也可以将结果输出到文件（先删除 /tmp/out1，不然会报错同名文件已经存在），继续执行以下命令：

  1. scala> counts.writeAsText("/tmp/out1")
  2. res1: org.apache.flink.api.java.operators.DataSink[(String, Int)] = DataSink '<unnamed>' (TextOutputFormat (/tmp/out1) - UTF-8)
  3.  
  4. scala> benv.execute("batch test")
  5. res2: org.apache.flink.api.common.JobExecutionResult = org.apache.flink.api.common.JobExecutionResult@737652a9

查看 /tmp/out1 文件就能看到输出结果。

  1. ➜  flink-1.7.2 cat /tmp/out1
  2. (be,2)
  3. (is,1)
  4. (not,1)
  5. (or,1)
  6. (question,1)
  7. (that,1)
  8. (the,1)
  9. (to,2)

DataSteam

  1. scala> val textStreaming = senv.fromElements("To be, or not to be,--that is the question:--")
  2. textStreaming: org.apache.flink.streaming.api.scala.DataStream[String] = org.apache.flink.streaming.api.scala.DataStream@4970b93d
  3.  
  4. scala> val countsStreaming = textStreaming.flatMap { _.toLowerCase.split("\\W+") }.map { (_, 1) }.keyBy(0).sum(1)
  5. countsStreaming: org.apache.flink.streaming.api.scala.DataStream[(String, Int)] = org.apache.flink.streaming.api.scala.DataStream@6a478680
  6.  
  7. scala> countsStreaming.print()
  8. res3: org.apache.flink.streaming.api.datastream.DataStreamSink[(String, Int)] = org.apache.flink.streaming.api.datastream.DataStreamSink@42bfc11f
  9.  
  10. scala> senv.execute("Streaming Wordcount")
  11. (to,1)
  12. (be,1)
  13. (or,1)
  14. (not,1)
  15. (to,2)
  16. (be,2)
  17. (that,1)
  18. (is,1)
  19. (the,1)
  20. (question,1)
  21. res4: org.apache.flink.api.common.JobExecutionResult = org.apache.flink.api.common.JobExecutionResult@1878815a

对 DataStream 任务，print() 并不会触发任务的执行，需要显示调用 execute(“job name”) 才会执行任务。

TableAPI

在 Blink 开源版本里面，支持了 TableAPI 方式提交任务（可以用 btenv.sqlQuery 提交 SQL 查询），社区版本 Flink 1.8 会支持 TableAPI: https://issues.apache.org/jira/browse/FLINK-9555

3.3 SQL Client Beta

SQL Client 目前还只是测试版，处于开发阶段，只能用于 SQL 的原型验证，不推荐在生产环境使用。

3.3.1 基本用法

  1. ➜  flink-1.7.2 bin/start-cluster.sh
  2. Starting cluster.
  3. Starting standalonesession daemon on host zkb-MBP.local.
  4. Starting taskexecutor daemon on host zkb-MBP.local.
  5.  
  6. ➜  flink-1.7.2 ./bin/sql-client.sh embedded
  7. No default environment specified.
  8. Searching for '/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf/sql-client-defaults.yaml'...found.
  9. Reading default environment from: file:/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf/sql-client-defaults.yaml
  10. No session environment specified.
  11. Validating current environment...done.
  12. … …
  13.  
  14. Flink SQL> help;
  15. The following commands are available:
  16.  
  17. QUIT        Quits the SQL CLI client.
  18. CLEAR       Clears the current terminal.
  19. HELP        Prints the available commands.
  20. SHOW TABLES     Shows all registered tables.
  21. SHOW FUNCTIONS      Shows all registered user-defined functions.
  22. DESCRIBE        Describes the schema of a table with the given name.
  23. EXPLAIN     Describes the execution plan of a query or table with the given name.
  24. SELECT      Executes a SQL SELECT query on the Flink cluster.
  25. INSERT INTO     Inserts the results of a SQL SELECT query into a declared table sink.
  26. CREATE VIEW     Creates a virtual table from a SQL query. Syntax: 'CREATE VIEW <name> AS <query>;'
  27. DROP VIEW       Deletes a previously created virtual table. Syntax: 'DROP VIEW <name>;'
  28. SOURCE      Reads a SQL SELECT query from a file and executes it on the Flink cluster.
  29. SET     Sets a session configuration property. Syntax: 'SET <key>=<value>;'. Use 'SET;' for listing all properties.
  30. RESET       Resets all session configuration properties.
  31.  
  32. Hint: Make sure that a statement ends with ';' for finalizing (multi-line) statements.

Select 查询

Flink SQL> SELECT 'Hello World';

按 ”Q” 退出这个界面 
打开 http://127.0.0.1:8081 能看到这条 Select 语句产生的查询任务已经结束了。这个查询采用的是读取固定数据集的 Custom Source，输出用的是 Stream Collect Sink，且只输出一条结果。

注意：如果本机的临时目录存在类似 .yarn-properties-baoniu 的文件，任务会提交到 Yarn 上。

Explain

Explain 命令可以查看 SQL 的执行计划。

  1. Flink SQL> explain SELECT name, COUNT(*) AS cnt FROM (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name) GROUP BY name;
  2.  
  3. == Abstract Syntax Tree ==        // 抽象语法树
  4. LogicalAggregate(group=[{0}], cnt=[COUNT()])
  5.   LogicalValues(tuples=[[{ _UTF-16LE'Bob  ' }, { _UTF-16LE'Alice' }, { _UTF-16LE'Greg ' }, { _UTF-16LE'Bob  ' }]])
  6.  
  7. == Optimized Logical Plan ==     // 优化后的逻辑执行计划
  8. DataStreamGroupAggregate(groupBy=[name], select=[name, COUNT(*) AS cnt])
  9.   DataStreamValues(tuples=[[{ _UTF-16LE'Bob  ' }, { _UTF-16LE'Alice' }, { _UTF-16LE'Greg ' }, { _UTF-16LE'Bob  ' }]])
  10.  
  11. == Physical Execution Plan ==   // 物理执行计划
  12. Stage 3 : Data Source
  13.     content : collect elements with CollectionInputFormat
  14.  
  15.     Stage 5 : Operator
  16.         content : groupBy: (name), select: (name, COUNT(*) AS cnt)
  17.         ship_strategy : HASH

3.3.2 结果展示

SQL Client 支持两种模式来维护并展示查询结果：

  • table mode: 在内存中物化查询结果，并以分页 table 形式展示。用户可以通过以下命令启用 table mode;

 SET execution.result-mode=table

  • changlog mode: 不会物化查询结果，而是直接对 continuous query 产生的添加和撤回（retractions）结果进行展示。

SET execution.result-mode=changelog

接下来通过实际的例子进行演示。

Table mode

  1. Flink SQL> SET execution.result-mode=table;
  2. [INFO] Session property has been set.
  3.  
  4. Flink SQL> SELECT name, COUNT(*) AS cnt FROM (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name) GROUP BY name;

运行结果如下图所示：

Changlog mode

  1. Flink SQL> SET execution.result-mode=changelog;
  2. [INFO] Session property has been set.
  3.  
  4. Flink SQL> SELECT name, COUNT(*) AS cnt FROM (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name) GROUP BY name;

运行结果如下图所示：

其中 ‘-’ 代表的就是撤回消息。

3.3.3 Environment Files

目前的 SQL Client 还不支持 DDL 语句，只能通过 yaml 文件的方式来定义 SQL 查询需要的表，UDF 和运行参数等信息。

首先，准备 env.yaml 和 input.csv 两个文件。

  1. ➜  flink-1.7.2 cat /tmp/env.yaml
  2. tables:
  3.   - name: MyTableSource
  4.     type: source-table
  5.     update-mode: append
  6.     connector:
  7.       type: filesystem
  8.       path: "/tmp/input.csv"
  9.     format:
  10.       type: csv
  11.       fields:
  12.         - name: MyField1
  13.           type: INT
  14.         - name: MyField2
  15.           type: VARCHAR
  16.       line-delimiter: "\n"
  17.       comment-prefix: "#"
  18.     schema:
  19.       - name: MyField1
  20.         type: INT
  21.       - name: MyField2
  22.         type: VARCHAR
  23.   - name: MyCustomView
  24.     type: view
  25.     query: "SELECT MyField2 FROM MyTableSource"
  26.   - name: MyTableSink
  27.     type: sink-table
  28.     update-mode: append
  29.     connector:
  30.       type: filesystem
  31.       path: "/tmp/output.csv"
  32.     format:
  33.       type: csv
  34.       fields:
  35.         - name: MyField1
  36.           type: INT
  37.         - name: MyField2
  38.           type: VARCHAR
  39.     schema:
  40.       - name: MyField1
  41.         type: INT
  42.       - name: MyField2
  43.         type: VARCHAR
  44.  
  45. # Execution properties allow for changing the behavior of a table program.
  46.  
  47. execution:
  48.   type: streaming                   # required: execution mode either 'batch' or 'streaming'
  49.   result-mode: table                # required: either 'table' or 'changelog'
  50.   max-table-result-rows: 1000000    # optional: maximum number of maintained rows in
  51.                                     #   'table' mode (1000000 by default, smaller 1 means unlimited)
  52.   time-characteristic: event-time   # optional: 'processing-time' or 'event-time' (default)
  53.   parallelism: 1                    # optional: Flink's parallelism (1 by default)
  54.   periodic-watermarks-interval: 200 # optional: interval for periodic watermarks (200 ms by default)
  55.   max-parallelism: 16               # optional: Flink's maximum parallelism (128 by default)
  56.   min-idle-state-retention: 0       # optional: table program's minimum idle state time
  57.   max-idle-state-retention: 0       # optional: table program's maximum idle state time
  58.   restart-strategy:                 # optional: restart strategy
  59.     type: fallback                  #   "fallback" to global restart strategy by default
  60.  
  61. # Deployment properties allow for describing the cluster to which table programs are submitted to.
  62.  
  63. deployment:
  64.   response-timeout: 5000
  65.  
  66. ➜  flink-1.7.2 cat /tmp/input.csv
  67. 1,hello
  68. 2,world
  69. 3,hello world
  70. 1,ok
  71. 3,bye bye
  72. 4,yes

启动 SQL Client：

  1. ➜  flink-1.7.2 ./bin/sql-client.sh embedded -e /tmp/env.yaml
  2. No default environment specified.
  3. Searching for '/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf/sql-client-defaults.yaml'...found.
  4. Reading default environment from: file:/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf/sql-client-defaults.yaml
  5. Reading session environment from: file:/tmp/env.yaml
  6. Validating current environment...done.
  7.  
  8. Flink SQL> show tables;
  9. MyCustomView
  10. MyTableSink
  11. MyTableSource
  12.  
  13. Flink SQL> describe MyTableSource;
  14. root
  15.  |-- MyField1: Integer
  16.  |-- MyField2: String
  17.  
  18. Flink SQL> describe MyCustomView;
  19. root
  20.  |-- MyField2: String
  21.  
  22. Flink SQL> create view MyView1 as select MyField1 from MyTableSource;
  23. [INFO] View has been created.
  24.  
  25. Flink SQL> show tables;
  26. MyCustomView
  27. MyTableSource
  28. MyView1
  29.  
  30. Flink SQL> describe MyView1;
  31. root
  32.  |-- MyField1: Integer
  33.  
  34. Flink SQL> select * from MyTableSource;

使用 insert into 写入结果表：

  1. Flink SQL> insert into MyTableSink select * from MyTableSource;
  2. [INFO] Submitting SQL update statement to the cluster...
  3. [INFO] Table update statement has been successfully submitted to the cluster:
  4. Cluster ID: StandaloneClusterId
  5. Job ID: 3fac2be1fd891e3e07595c684bb7b7a0
  6. Web interface: http://localhost:8081

查询生成的结果数据文件：

  1. ➜  flink-1.7.2 cat /tmp/output.csv
  2. 1,hello
  3. 2,world
  4. 3,hello world
  5. 1,ok
  6. 3,bye bye
  7. 4,yes

也可以在 Environment 文件里面定义 UDF，在 SQL Client 里面通过 「HOW FUNCTIONS」查询和使用，这里就不再说明了。

SQL Client 功能社区还在开发中，详见 FLIP-24。

3.4 Restful API

接下来我们演示如何通过 Rest API 来提交 Jar 包和执行任务。

更详细的操作请参考 Flink 的 Restful API 文档：https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/rest_api.html

  1. ➜  flink-1.7.2 curl http://127.0.0.1:8081/overview
  2. {"taskmanagers":1,"slots-total":4,"slots-available":0,"jobs-running":3,"jobs-finished":0,"jobs-cancelled":0,"jobs-failed":0,"flink-version":"1.7.2","flink-commit":"ceba8af"}%
  3.  
  4. ➜  flink-1.7.2 curl -X POST -H "Expect:" -F "jarfile=@/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/examples/streaming/TopSpeedWindowing.jar" http://127.0.0.1:8081/jars/upload
  5. {"filename":"/var/folders/2b/r6d49pcs23z43b8fqsyz885c0000gn/T/flink-web-124c4895-cf08-4eec-8e15-8263d347efc2/flink-web-upload/6077eca7-6db0-4570-a4d0-4c3e05a5dc59_TopSpeedWindowing.jar","status":"success"}%       
  6.                                                                                                                                                                                                    ➜  flink-1.7.2 curl http://127.0.0.1:8081/jars
  7. {"address":"http://localhost:8081","files":[{"id":"6077eca7-6db0-4570-a4d0-4c3e05a5dc59_TopSpeedWindowing.jar","name":"TopSpeedWindowing.jar","uploaded":1553743438000,"entry":[{"name":"org.apache.flink.streaming.examples.windowing.TopSpeedWindowing","description":null}]}]}%
  8.  
  9. ➜  flink-1.7.2 curl http://127.0.0.1:8081/jars/6077eca7-6db0-4570-a4d0-4c3e05a5dc59_TopSpeedWindowing.jar/plan
  10. {"plan":{"jid":"41029eb3feb9132619e454ec9b2a89fb","name":"CarTopSpeedWindowingExample","nodes":[{"id":"90bea66de1c231edf33913ecd54406c1","parallelism":1,"operator":"","operator_strategy":"","description":"Window(GlobalWindows(), DeltaTrigger, TimeEvictor, ComparableAggregator, PassThroughWindowFunction) -> Sink: Print to Std. Out","inputs":[{"num":0,"id":"cbc357ccb763df2852fee8c4fc7d55f2","ship_strategy":"HASH","exchange":"pipelined_bounded"}],"optimizer_properties":{}},{"id":"cbc357ccb763df2852fee8c4fc7d55f2","parallelism":1,"operator":"","operator_strategy":"","description":"Source: Custom Source -> Timestamps/Watermarks","optimizer_properties":{}}]}}%                                                                                                                                                    ➜  flink-1.7.2 curl -X POST http://127.0.0.1:8081/jars/6077eca7-6db0-4570-a4d0-4c3e05a5dc59_TopSpeedWindowing.jar/run
  11. {"jobid":"04d80a24b076523d3dc5fbaa0ad5e1ad"}%

Restful API 还提供了很多监控和 Metrics 相关的功能，对于任务提交的操作也支持的比较全面。

3.5 Web

在 Flink Dashboard 页面左侧可以看到有个「Submit new Job」的地方，用户可以上传 Jar 包和显示执行计划和提交任务。Web 提交功能主要用于新手入门和演示用。

4. 结语

本期的课程到这里就结束了，我们主要讲解了 Flink 的 5 种任务提交的方式。熟练掌握各种任务提交方式，有利于提高我们日常的开发和运维效率。

文章来自：ververica

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/548710
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号