碼上歡樂
相關內容    簡體    繁體

Apache Flink 零基礎入門（四）：客戶端操作的 5 種模式

本文轉載自   查看原文   2020-03-02 22:59   725    flink/ Flink

來源於  https://developer.aliyun.com/article/712704

簡介： 本文主要分享 Flink 的 5 種任務提交的方式。熟練掌握各種任務提交方式，有利於提高我們日常的開發和運維效率。

作者：周凱波（寶牛）

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_clients.png

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

flink_1_7_2.jpg

3.Flink 客戶端操作

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 的集群：

  flink-1.7.2 bin/start-cluster.sh Starting cluster. Starting standalonesession daemon on host zkb-MBP.local. Starting taskexecutor daemon on host zkb-MBP.local.

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

Run

運行任務，以 Flink 自帶的例子 TopSpeedWindowing 為例：

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

運行起來后默認是 1 個並發:

flink_run_1.jpg

點左側「Task Manager」，然后點「Stdout」能看到輸出日志：

flink_run_2.jpg

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

flink_run_3.jpg

List

查看任務列表：

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

Stop

停止任務。通過 -m 來指定要停止的 JobManager 的主機地址和端口。

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

從日志里面能看出 Stop 命令執行失敗了。一個 Job 能夠被 Stop 要求所有的 Source 都是可以 Stoppable 的，即實現了 StoppableFunction 接口。

 
/**  * 需要能 stoppable 的函數必須實現這個接口，例如流式任務的 source。  * stop() 方法在任務收到 STOP 信號的時候調用。  * source 在接收到這個信號后，必須停止發送新的數據且優雅的停止。 */ @PublicEvolving public interface StoppableFunction { /**  * 停止 source。與 cancel() 不同的是，這是一個讓 source 優雅停止的請求。  * 等待中的數據可以繼續發送出去，不需要立即停止。  */ void stop(); }

Cancel

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

  flink-1.7.2 bin/flink cancel -m 127.0.0.1:8081 5e20cb6b0f357591171dfcca2eea09de Cancelling job 5e20cb6b0f357591171dfcca2eea09de. Cancelled job 5e20cb6b0f357591171dfcca2eea09de.

也可以在停止的時候顯示指定 Savepoint 目錄。

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

取消和停止（流作業）的區別如下：

  • cancel() 調用，立即調用作業算子的 cancel() 方法，以盡快取消它們。如果算子在接到 cancel() 調用后沒有停止，Flink 將開始定期中斷算子線程的執行，直到所有算子停止為止。
  • stop() 調用，是更優雅的停止正在運行流作業的方式。stop() 僅適用於 Source 實現了 StoppableFunction 接口的作業。當用戶請求停止作業時，作業的所有 Source 都將接收 stop() 方法調用。直到所有 Source 正常關閉時，作業才會正常結束。這種方式，使作業正常處理完所有作業。

Savepoint

觸發 Savepoint。

  flink-1.7.2 bin/flink savepoint -m 127.0.0.1:8081 ec53edcfaeb96b2a5dadbfbe5ff62bbb /tmp/savepoint Triggering savepoint for job ec53edcfaeb96b2a5dadbfbe5ff62bbb. Waiting for response... Savepoint completed. Path: file:/tmp/savepoint/savepoint-ec53ed-84b00ce500ee 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 啟動：

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

查看 JobManager 的日志，能夠看到類似這樣的 Log：

2019-03-28 10:30:53,957 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Starting job 790d7b98db6f6af55d04aec1d773852d from savepoint /tmp/savepoint/savepoint-f049ff-24ec0d3e0dc7 () 2019-03-28 10:30:53,959 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Reset the checkpoint ID of job 790d7b98db6f6af55d04aec1d773852d to 2. 2019-03-28 10:30:53,959 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - 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，提示無法識別）

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

修改參數后需要重啟集群生效，然后再啟動任務：

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

從頁面上能看到 Task Slot 變為了 4，這時候任務的默認並發度是 1。

standalone-modify-1.jpg

standalone-modify-2.jpg

通過 Modify 命令依次將並發度修改為 4 和 3，可以看到每次 Modify 命令都會觸發一次 Savepoint。

  flink-1.7.2 bin/flink modify -p 4 7752ea7b0e7303c780de9d86a5ded3fa Modify job 7752ea7b0e7303c780de9d86a5ded3fa. Rescaled job 7752ea7b0e7303c780de9d86a5ded3fa. Its new parallelism is 4. flink-1.7.2 ll /tmp/savepoint total 0 drwxr-xr-x 3 baoniu 96 Jun 17 09:05 savepoint-7752ea-00c05b015836/ flink-1.7.2 bin/flink modify -p 3 7752ea7b0e7303c780de9d86a5ded3fa Modify job 7752ea7b0e7303c780de9d86a5ded3fa. Rescaled job 7752ea7b0e7303c780de9d86a5ded3fa. Its new parallelism is 3. flink-1.7.2 ll /tmp/savepoint total 0 drwxr-xr-x 3 baoniu 96 Jun 17 09:08 savepoint-7752ea-449b131b2bd4/

standalone-modify-3.jpg

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

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 () 2019-06-17 09:05:11,182 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Reset the checkpoint ID of job 7752ea7b0e7303c780de9d86a5ded3fa to 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. 2019-06-17 09:05:11,184 INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - No master state to restore 2019-06-17 09:05:11,184 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job CarTopSpeedWindowingExample (7752ea7b0e7303c780de9d86a5ded3fa) switched from state RUNNING to SUSPENDING. org.apache.flink.util.FlinkException: Job is being rescaled.

Info

Info 命令是用來查看 Flink 任務的執行計划（StreamGraph）的。

  flink-1.7.2 bin/flink info examples/streaming/TopSpeedWindowing.jar ----------------------- Execution Plan ----------------------- {"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"}]}]} --------------------------------------------------------------

拷貝輸出的 Json 內容，粘貼到這個網站：http://flink.apache.org/visualizer/

visualizer.jpg

可以和實際運行的物理執行計划對比：

physical-execute-plan.jpg

3.1.2 Yarn per-job

單任務 Attach 模式

默認是 Attach 模式，即客戶端會一直等待直到程序結束才會退出。

  • 通過 -m yarn-cluster 指定 Yarn 模式
  • Yarn 上顯示名字為 Flink session cluster，這個 Batch 的 Wordcount 任務運行完會 FINISHED。
  • 客戶端能看到結果輸出
[admin@z17.sqa.zth /home/admin/flink/flink-1.7.2] $echo $HADOOP_CONF_DIR /etc/hadoop/conf/ [admin@z17.sqa.zth /home/admin/flink/flink-1.7.2] $./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar 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 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 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 2019-06-17 09:15:24,907 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=4} 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. 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. 2019-06-17 09:15:36,239 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application master application_1532332183347_0724 2019-06-17 09:15:36,276 INFO org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application application_1532332183347_0724 2019-06-17 09:15:36,276 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster to be allocated 2019-06-17 09:15:36,281 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster, current state ACCEPTED 2019-06-17 09:15:40,426 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has been deployed successfully. Starting execution of program Executing WordCount example with default input data set. Use --input to specify file input. Printing result to stdout. Use --output to specify output path. (a,5) (action,1) (after,1) (against,1) (all,2) ... ... (would,2) (wrong,1) (you,1) Program execution finished Job with JobID 8bfe7568cb5c3254af30cbbd9cd5971e has finished. Job Runtime: 9371 ms Accumulator Results: - 2bed2c5506e9237fb85625416a1bc508 (java.util.ArrayList) [170 elements]

yarn-1.jpg

yarn-2.jpg

如果我們以 Attach 模式運行 Streaming 的任務，客戶端會一直等待不退出，可以運行以下的例子試驗下：

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

單任務 Detached 模式

  • 由於是 Detached 模式，客戶端提交完任務就退出了
  • Yarn 上顯示為 Flink per-job cluster
$./bin/flink run -yd -m yarn-cluster ./examples/streaming/TopSpeedWindowing.jar 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 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 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 2019-06-18 09:21:59,940 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=4} 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. 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. ^@2019-06-18 09:22:12,113 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application master application_1532332183347_0729 2019-06-18 09:22:12,151 INFO org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application application_1532332183347_0729 2019-06-18 09:22:12,151 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster to be allocated 2019-06-18 09:22:12,155 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster, current state ACCEPTED 2019-06-18 09:22:16,275 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has been deployed successfully. 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: yarn application -kill application_1532332183347_0729 Please also note that the temporary files of the YARN session in the home directory will not be removed. Job has been submitted with JobID e61b9945c33c300906ad50a9a11f36df

yarn-detached-1.jpg

yarn-detached-2.jpg

3.1.3 Yarn session

啟動 Session

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

表示啟動一個 Yarn session 集群，每個 TM 的內存是 2 G，每個 TM 有 3 個 Slot。(注意：-n 參數不生效)

  flink-1.7.2 ./bin/yarn-session.sh -tm 2048 -s 3 2019-06-17 09:21:50,177 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.address, localhost 2019-06-17 09:21:50,179 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.port, 6123 2019-06-17 09:21:50,179 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.heap.size, 1024m 2019-06-17 09:21:50,179 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.heap.size, 1024m 2019-06-17 09:21:50,179 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.numberOfTaskSlots, 4 2019-06-17 09:21:50,179 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: state.savepoints.dir, file:///tmp/savepoint 2019-06-17 09:21:50,180 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: parallelism.default, 1 2019-06-17 09:21:50,180 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: rest.port, 8081 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 2019-06-17 09:21:50,746 INFO org.apache.flink.runtime.security.modules.HadoopModule - Hadoop user set to baoniu (auth:SIMPLE) 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 2019-06-17 09:21:51,148 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=2048, numberTaskManagers=1, slotsPerTaskManager=3} 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. 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. ^@2019-06-17 09:22:03,304 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application master application_1532332183347_0726 2019-06-17 09:22:03,336 INFO org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application application_1532332183347_0726 2019-06-17 09:22:03,336 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster to be allocated 2019-06-17 09:22:03,340 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster, current state ACCEPTED 2019-06-17 09:22:07,722 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has been deployed successfully. 2019-06-17 09:22:08,050 INFO org.apache.flink.runtime.rest.RestClient - Rest client endpoint started. Flink JobManager is now running on z07.sqa.net:37109 with leader id 00000000-0000-0000-0000-000000000000. 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；

yarn-session-1.jpg

yarn-session-2.jpg

  • 在本機的臨時目錄（有些機器是 /tmp 目錄）下會生成一個文件：
  flink-1.7.2 cat /var/folders/2b/r6d49pcs23z43b8fqsyz885c0000gn/T/.yarn-properties-baoniu #Generated YARN properties file #Mon Jun 17 09:22:08 CST 2019 parallelism=3 dynamicPropertiesString= applicationID=application_1532332183347_0726

提交任務

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

將會根據 /tmp/.yarn-properties-admin 文件內容提交到了剛啟動的 Session。

  flink-1.7.2 ./bin/flink run ./examples/batch/WordCount.jar 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. 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. 2019-06-17 09:26:43,058 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - YARN properties set default parallelism to 3 2019-06-17 09:26:43,058 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - YARN properties set default parallelism to 3 YARN properties set default parallelism to 3 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 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 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 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' Starting execution of program Executing WordCount example with default input data set. Use --input to specify file input. Printing result to stdout. Use --output to specify output path. ^@(a,5) (action,1) (after,1) (against,1) (all,2) (and,12) ... ... (wrong,1) (you,1) Program execution finished Job with JobID ad9b0f1feed6d0bf6ba4e0f18b1e65ef has finished. Job Runtime: 9152 ms Accumulator Results: - fd07c75d503d0d9a99e4f27dd153114c (java.util.ArrayList) [170 elements]

運行結束后 TM 的資源會釋放。

session-sumit-job1.jpg

提交到指定的 Session

通過 -yid 參數來提交到指定的 Session。

$./bin/flink run -d -p 30 -m yarn-cluster -yid application_1532332183347_0708 ./examples/streaming/TopSpeedWindowing.jar 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 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 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 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' Starting execution of program Executing TopSpeedWindowing example with default input data set. Use --input to specify file input. Printing result to stdout. Use --output to specify output path. Job has been submitted with JobID 58d5049ebbf28d515159f2f88563f5fd

yarn-specifiy-session-1.jpg

注：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

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

任務運行說明：

  • Batch 任務內置了 benv 變量，通過 print() 將結果輸出到控制台；
  • Streaming 任務內置了 senv 變量，通過 senv.execute("job name") 來提交任務，且 Datastream 的輸出只有在 Local 模式下打印到控制台；

Remote

先啟動一個 yarn session cluster：

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

啟動 scala shell，連到 jm：

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

Yarn

$./bin/start-scala-shell.sh yarn -n 2 -jm 1024 -s 2 -tm 1024 -nm flink-yarn Starting Flink Shell: 2019-03-25 09:47:44,695 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.address, localhost 2019-03-25 09:47:44,697 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.rpc.port, 6123 2019-03-25 09:47:44,697 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: jobmanager.heap.size, 1024m 2019-03-25 09:47:44,697 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.heap.size, 1024m 2019-03-25 09:47:44,697 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: taskmanager.numberOfTaskSlots, 4 2019-03-25 09:47:44,698 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: parallelism.default, 1 2019-03-25 09:47:44,698 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: state.savepoints.dir, file:///tmp/savepoint 2019-03-25 09:47:44,698 INFO org.apache.flink.configuration.GlobalConfiguration - Loading configuration property: rest.port, 8081 2019-03-25 09:47:44,717 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - Found Yarn properties file under /tmp/.yarn-properties-admin. 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 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 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 2019-03-25 09:47:45,275 INFO org.apache.flink.yarn.cli.FlinkYarnSessionCli - The argument yn is deprecated in will be ignored. 2019-03-25 09:47:45,357 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=2, slotsPerTaskManager=2} 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. 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. 2019-03-25 09:47:46,514 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Submitting application master application_1532332183347_0710 2019-03-25 09:47:46,534 INFO org.apache.hadoop.yarn.client.api.impl.YarnClientImpl - Submitted application application_1532332183347_0710 2019-03-25 09:47:46,534 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Waiting for the cluster to be allocated 2019-03-25 09:47:46,535 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - Deploying cluster, current state ACCEPTED 2019-03-25 09:47:51,051 INFO org.apache.flink.yarn.AbstractYarnClusterDescriptor - YARN application has been deployed successfully. 2019-03-25 09:47:51,222 INFO org.apache.flink.runtime.rest.RestClient - Rest client endpoint started. Connecting to Flink cluster (host: 10.10.10.10, port: 56942).

scala-shell-yarn.jpg

按 CTRL + C 退出 Shell 后，這個 Flink cluster 還會繼續運行，不會退出。

3.2.2 Execute

DataSet

  flink-1.7.2 bin/stop-cluster.sh No taskexecutor daemon to stop on host zkb-MBP.local. No standalonesession daemon to stop on host zkb-MBP.local. flink-1.7.2 bin/start-scala-shell.sh local Starting Flink Shell: Starting local Flink cluster (host: localhost, port: 8081). Connecting to Flink cluster (host: localhost, port: 8081). scala> val text = benv.fromElements("To be, or not to be,--that is the question:--") text: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@5b407336 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@6ee34fe4 scala> counts.print() (be,2) (is,1) (not,1) (or,1) (question,1) (that,1) (the,1) (to,2)

對 DataSet 任務來說，print() 會觸發任務的執行。

scala-shell-dataset-run-1.jpg

也可以將結果輸出到文件（先刪除 /tmp/out1，不然會報錯同名文件已經存在），繼續執行以下命令：

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

查看 /tmp/out1 文件就能看到輸出結果。

  flink-1.7.2 cat /tmp/out1 (be,2) (is,1) (not,1) (or,1) (question,1) (that,1) (the,1) (to,2)

scala-shell-dataset-run-2.jpg

DataSteam

scala> val textStreaming = senv.fromElements("To be, or not to be,--that is the question:--") textStreaming: org.apache.flink.streaming.api.scala.DataStream[String] = org.apache.flink.streaming.api.scala.DataStream@4970b93d scala> val countsStreaming = textStreaming.flatMap { _.toLowerCase.split("\\W+") }.map { (_, 1) }.keyBy(0).sum(1) countsStreaming: org.apache.flink.streaming.api.scala.DataStream[(String, Int)] = org.apache.flink.streaming.api.scala.DataStream@6a478680 scala> countsStreaming.print() res3: org.apache.flink.streaming.api.datastream.DataStreamSink[(String, Int)] = org.apache.flink.streaming.api.datastream.DataStreamSink@42bfc11f scala> senv.execute("Streaming Wordcount") (to,1) (be,1) (or,1) (not,1) (to,2) (be,2) (that,1) (is,1) (the,1) (question,1) res4: org.apache.flink.api.common.JobExecutionResult = org.apache.flink.api.common.JobExecutionResult@1878815a

對 DataStream 任務，print() 並不會觸發任務的執行，需要顯示調用 execute(“job name”) 才會執行任務。

scala-shell-datastream-1.jpg

scala-shell-datastream-2.jpg

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 基本用法

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

Select 查詢

Flink SQL> SELECT 'Hello World';

sql-select.jpg

按 ”Q” 退出這個界面
打開 http://127.0.0.1:8081 能看到這條 Select 語句產生的查詢任務已經結束了。這個查詢采用的是讀取固定數據集的 Custom Source，輸出用的是 Stream Collect Sink，且只輸出一條結果。

注意：如果本機的臨時目錄存在類似 .yarn-properties-baoniu 的文件，任務會提交到 Yarn 上。

sql-select-run-1.jpg

sql-select-run-2.jpg

Explain

Explain 命令可以查看 SQL 的執行計划。

Flink SQL> explain SELECT name, COUNT(*) AS cnt FROM (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name) GROUP BY name; == Abstract Syntax Tree == // 抽象語法樹 LogicalAggregate(group=[{0}], cnt=[COUNT()]) LogicalValues(tuples=[[{ _UTF-16LE'Bob ' }, { _UTF-16LE'Alice' }, { _UTF-16LE'Greg ' }, { _UTF-16LE'Bob ' }]]) == Optimized Logical Plan == // 優化后的邏輯執行計划 DataStreamGroupAggregate(groupBy=[name], select=[name, COUNT(*) AS cnt]) DataStreamValues(tuples=[[{ _UTF-16LE'Bob ' }, { _UTF-16LE'Alice' }, { _UTF-16LE'Greg ' }, { _UTF-16LE'Bob ' }]]) == Physical Execution Plan == // 物理執行計划 Stage 3 : Data Source content : collect elements with CollectionInputFormat Stage 5 : Operator content : groupBy: (name), select: (name, COUNT(*) AS cnt) 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

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

運行結果如下圖所示：

sql-table-mode-1.jpg

sql-table-mode-2.jpg

sql-table-mode-3.jpg

Changlog mode

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

運行結果如下圖所示：

sql-change-mode-1.jpg

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

sql-change-mode-2.jpg

sql-change-mode-3.jpg

3.3.3 Environment Files

目前的 SQL Client 還不支持 DDL 語句，只能通過 yaml 文件的方式來定義 SQL 查詢需要的表，UDF 和運行參數等信息。

首先，准備 env.yaml 和 input.csv 兩個文件。

  flink-1.7.2 cat /tmp/env.yaml
tables:
  - name: MyTableSource
    type: source-table
    update-mode: append connector: type: filesystem path: "/tmp/input.csv" format: type: csv fields: - name: MyField1 type: INT - name: MyField2 type: VARCHAR line-delimiter: "\n" comment-prefix: "#" schema: - name: MyField1 type: INT - name: MyField2 type: VARCHAR - name: MyCustomView type: view query: "SELECT MyField2 FROM MyTableSource" - name: MyTableSink type: sink-table update-mode: append connector: type: filesystem path: "/tmp/output.csv" format: type: csv fields: - name: MyField1 type: INT - name: MyField2 type: VARCHAR schema: - name: MyField1 type: INT - name: MyField2 type: VARCHAR # Execution properties allow for changing the behavior of a table program. execution: type: streaming # required: execution mode either 'batch' or 'streaming' result-mode: table # required: either 'table' or 'changelog' max-table-result-rows: 1000000 # optional: maximum number of maintained rows in # 'table' mode (1000000 by default, smaller 1 means unlimited) time-characteristic: event-time # optional: 'processing-time' or 'event-time' (default) parallelism: 1 # optional: Flink's parallelism (1 by default) periodic-watermarks-interval: 200 # optional: interval for periodic watermarks (200 ms by default) max-parallelism: 16 # optional: Flink's maximum parallelism (128 by default) min-idle-state-retention: 0 # optional: table program's minimum idle state time max-idle-state-retention: 0 # optional: table program's maximum idle state time restart-strategy: # optional: restart strategy type: fallback # "fallback" to global restart strategy by default # Deployment properties allow for describing the cluster to which table programs are submitted to. deployment: response-timeout: 5000 flink-1.7.2 cat /tmp/input.csv 1,hello 2,world 3,hello world 1,ok 3,bye bye 4,yes

啟動 SQL Client：

  flink-1.7.2 ./bin/sql-client.sh embedded -e /tmp/env.yaml
No default environment specified. Searching for '/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf/sql-client-defaults.yaml'...found. Reading default environment from: file:/Users/baoniu/Documents/work/tool/flink/flink-1.7.2/conf/sql-client-defaults.yaml Reading session environment from: file:/tmp/env.yaml Validating current environment...done. Flink SQL> show tables; MyCustomView MyTableSink MyTableSource Flink SQL> describe MyTableSource; root |-- MyField1: Integer |-- MyField2: String Flink SQL> describe MyCustomView; root |-- MyField2: String Flink SQL> create view MyView1 as select MyField1 from MyTableSource; [INFO] View has been created. Flink SQL> show tables; MyCustomView MyTableSource MyView1 Flink SQL> describe MyView1; root |-- MyField1: Integer Flink SQL> select * from MyTableSource;

sql_env_file_1.jpg

sql_env_file_2.jpg

sql_env_file_3.jpg

使用 insert into 寫入結果表：

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

sql_insert_into_1.jpg

sql_insert_into_2.jpg

查詢生成的結果數據文件：

  flink-1.7.2 cat /tmp/output.csv 1,hello 2,world 3,hello world 1,ok 3,bye bye 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

  flink-1.7.2 curl http://127.0.0.1:8081/overview {"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"}% 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 {"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"}% flink-1.7.2 curl http://127.0.0.1:8081/jars {"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}]}]}% flink-1.7.2 curl http://127.0.0.1:8081/jars/6077eca7-6db0-4570-a4d0-4c3e05a5dc59_TopSpeedWindowing.jar/plan {"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 {"jobid":"04d80a24b076523d3dc5fbaa0ad5e1ad"}%

rest_api_1.jpg

rest_api_2.jpg

Restful API 還提供了很多監控和 Metrics 相關的功能，對於任務提交的操作也支持的比較全面。

3.5 Web

在 Flink Dashboard 頁面左側可以看到有個「Submit new Job」的地方，用戶可以上傳 Jar 包和顯示執行計划和提交任務。Web 提交功能主要用於新手入門和演示用。

flink_web_1.jpg

4.結束

本期的課程到這里就結束了，我們主要講解了 Flink 的 5 種任務提交的方式。熟練掌握各種任務提交方式，有利於提高我們日常的開發和運維效率。

視頻回顧：https://zh.ververica.com/developers/flink-training-course2/


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 Apache Flink 零基礎入門（七）Flink中keyBy三種方式指定key apache sshd客戶端 Akka（37）： Http：客戶端操作模式 es6.6.1 java客戶端 client基礎操作 hdfs的客戶端操作 ElasticSearch的客戶端操作 HDFS的客戶端操作 etcdctl客戶端操作 SVN客戶端操作 29-ESP8266 SDK開發基礎入門篇--編寫TCP 客戶端程序(Lwip RAW模式,非RTOS版,精簡入門)
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM