Flink 客戶端操作命令及可視化工具

Flink提供了豐富的客戶端操作來提交任務和與任務進行交互。下面主要從Flink命令行、Scala Shell、SQL Client、Restful API和 Web五個方面進行整理。

在Flink安裝目錄的bin目錄下可以看到flink，start-scala-shell.sh和sql-client.sh等文件，這些都是客戶端操作的入口。
[點擊并拖拽以移動] ?

flink 常見操作：可以通過 -help 查看幫助

run 運行任務

-d：以分離模式運行作業
-c：如果沒有在jar包中指定入口類，則需要在這里通過這個參數指定；
-m：指定需要連接的jobmanager(主節點)地址，使用這個參數可以指定一個不同于配置文件中的jobmanager，可以說是yarn集群名稱；
-p：指定程序的并行度。可以覆蓋配置文件中的默認值；
-s：保存點savepoint的路徑以還原作業來自（例如hdfs:///flink/savepoint-1537)；

[root@hadoop1 flink-1.10.1]# bin/flink run -d examples/streaming/TopSpeedWindowing.jar 
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 dce7b69ad15e8756766967c46122736f

就可以看到我們提交的JobManager，默認是一個并發。
[點擊并拖拽以移動] ?

點進去就可以看到詳細的信息
[點擊并拖拽以移動] ?

點擊左側TaskManager —Stdout能看到具體輸出的日志信息。
[點擊并拖拽以移動] ?

或者查看TaskManager節點的log目錄下的*.out文件，也能看到具體的輸出信息。
[點擊并拖拽以移動] ?

list 查看任務列表

-m：jobmanager<arg>作業管理器（主）的地址連接。

[root@hadoop1 flink-1.10.1]# bin/flink list -m 127.0.0.1:8081
Waiting for response...
------------------ Running/Restarting Jobs -------------------
09.07.2020 16:44:09 : dce7b69ad15e8756766967c46122736f : CarTopSpeedWindowingExample (RUNNING)
--------------------------------------------------------------
No scheduled jobs.

Stop 停止任務

需要指定jobmanager的ip:prot和jobId。如下報錯可知，一個job能夠被stop要求所有的source都是可以stoppable的，即實現了 StoppableFunction接口。

[root@hadoop1 flink-1.10.1]# bin/flink stop -m 127.0.0.1:8081 dce7b69ad15e8756766967c46122736f
Suspending job "dce7b69ad15e8756766967c46122736f" with a savepoint.------------------------------------------------------------The program finished with the following exception:org.apache.flink.util.FlinkException: Could not stop with a savepoint job "dce7b69ad15e8756766967c46122736f".at org.apache.flink.client.cli.CliFrontend.lambda$stop$5(CliFrontend.java:458)

StoppableFunction接口如下，屬于優雅停止任務。

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

Cancel 取消任務

如果在conf/flink-conf.yaml里面配置state.savepoints.dir，會保存savepoint，否則不會保存savepoint。（重啟）

state.savepoints.dir: file:///tmp/savepoint

執行 Cancel命令 取消任務

[root@hadoop1 flink-1.10.1]# bin/flink cancel -m 127.0.0.1:8081 -s e8ce0d111262c52bf8228d5722742d47
DEPRECATION WARNING: Cancelling a job with savepoint is deprecated. Use "stop" instead.
Cancelling job e8ce0d111262c52bf8228d5722742d47 with savepoint to default savepoint directory.
Cancelled job e8ce0d111262c52bf8228d5722742d47. Savepoint stored in file:/tmp/savepoint/savepoint-e8ce0d-f7fa96a085d8.

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

1 [root@hadoop1 flink-1.10.1]# bin/flink cancel -m 127.0.0.1:8081 -s /tmp/savepoint f58bb4c49ee5580ab5f27fdb24083353
DEPRECATION WARNING: Cancelling a job with savepoint is deprecated. Use "stop" instead.
Cancelling job f58bb4c49ee5580ab5f27fdb24083353 with savepoint to /tmp/savepoint.
Cancelled job f58bb4c49ee5580ab5f27fdb24083353. Savepoint stored in file:/tmp/savepoint/savepoint-f58bb4-127b7e84910e.

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

觸發 savepoint

當需要生成savepoint文件時，需要手動觸發savepoint。如下，需要指定正在運行的 JobID 和生成文件的存放目錄。同時，我們也可以看到它會返回給用戶存放的savepoint的文件名稱等信息。

 [root@hadoop1 flink-1.10.1]# bin/flink run -d examples/streaming/TopSpeedWindowing.jar 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 216c427d63e3754eb757d2cc268a448d[root@hadoop1 flink-1.10.1]# bin/flink savepoint -m 127.0.0.1:8081 216c427d63e3754eb757d2cc268a448d /tmp/savepoint/Triggering savepoint for job 216c427d63e3754eb757d2cc268a448d.Waiting for response...Savepoint completed. Path: file:/tmp/savepoint/savepoint-216c42-154a34cf6bfdYou can resume your program from this savepoint with the run command.

savepoint和checkpoint的區別：
● checkpoint是增量做的，每次的時間較短，數據量較小，只要在程序里面啟用后會自動觸發，用戶無須感知；savepoint是全量做的，每次的時間較長，數據量較大，需要用戶主動去觸發。
● checkpoint是作業failover的時候自動使用，不需要用戶指定。savepoint一般用于程序的版本更新，bug修復，A/B Test等場景，需要用戶指定。

從指定 savepoint 中啟動

[root@hadoop1 flink-1.10.1]# bin/flink run -d -s /tmp/savepoint/savepoint-f58bb4-127b7e84910e/ examples/streaming/TopSpeedWindowing.jar 
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 1a5c5ce279e0e4bd8609f541b37652e2

查看JobManager的日志能夠看到Reset the checkpoint ID為我們指定的savepoint文件中的ID
[點擊并拖拽以移動] ?

modify 修改任務并行度

這里修改master的conf/flink-conf.yaml將task slot數修改為4。并通過xsync分發到兩個slave節點上。

taskmanager.numberOfTaskSlots: 4

修改參數后需要重啟集群生效：關閉/啟動集群

[root@hadoop1 flink-1.10.1]# bin/stop-cluster.sh && bin/start-cluster.sh 
Stopping taskexecutor daemon (pid: 8236) on host hadoop2.
Stopping taskexecutor daemon (pid: 8141) on host hadoop3.
Stopping standalonesession daemon (pid: 22633) on host hadoop1.
Starting cluster.
Starting standalonesession daemon on host hadoop1.
Starting taskexecutor daemon on host hadoop2.
Starting taskexecutor daemon on host hadoop3.

啟動任務

[root@hadoop1 flink-1.10.1]# bin/flink run -d examples/streaming/TopSpeedWindowing.jar 
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 2e833a438da7d8052f14d5433910515a

從頁面上能看到Task Slots總計變為了8，運行的Slot為1，剩余Slot數量為7。
[點擊并拖拽以移動] ?

這時候默認的并行度是1
[點擊并拖拽以移動] ?

Flink1.0版本命令行flink modify已經沒有這個行為了，被移除了。。。Flink1.7上是可以運行的。

[root@hadoop1 flink-1.10.1]# bin/flink modify -p 4 cc22cc3d09f5d65651d637be6fb0a1c3
"modify" is not a valid action.

Info 顯示程序的執行計劃

[root@hadoop1 flink-1.10.1]# 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/可以生成類似如下的執行圖。

[點擊并拖拽以移動] ?

可以與實際運行的物理執行計劃進行對比。
[點擊并拖拽以移動] ?

SQL Client Beta

進入 Flink SQL

[root@hadoop1 flink-1.10.1]# bin/sql-client.sh embedded

Select查詢，按Q退出如下界面；

Flink SQL> select 'hello word';SQL Query Result (Table)Table program finished.                                                                                       Page: Last of 1                                                                                         Updated: 16:37:04.649EXPR$0hello wordQ Quit                                         + Inc Refresh                                  G Goto Page                                    N Next Page                                    O Open Row
R Refresh                                      - Dec Refresh                                  L Last Page                                    P Prev Page

打開http://hadoop1:8081能看到這條select語句產生的查詢任務已經結束了。這個查詢采用的是讀取固定數據集的Custom Source，輸出用的是Stream Collect Sink，且只輸出一條結果。
[點擊并拖拽以移動] ?

[點擊并拖拽以移動] ?

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(type=[RecordType(VARCHAR(5) name)], tuples=[[{ _UTF-16LE'Bob' }, { _UTF-16LE'Alice' }, { _UTF-16LE'Greg' }, { _UTF-16LE'Bob' }]])== Optimized Logical Plan ==      //優化后的邏輯執行計劃
GroupAggregate(groupBy=[name], select=[name, COUNT(*) AS cnt])
+- Exchange(distribution=[hash[name]])+- Values(type=[RecordType(VARCHAR(5) name)], tuples=[[{ _UTF-16LE'Bob' }, { _UTF-16LE'Alice' }, { _UTF-16LE'Greg' }, { _UTF-16LE'Bob' }]])== Physical Execution Plan ==    //物理執行計劃
Stage 13 : Data Sourcecontent : Source: Values(tuples=[[{ _UTF-16LE'Bob' }, { _UTF-16LE'Alice' }, { _UTF-16LE'Greg' }, { _UTF-16LE'Bob' }]])Stage 15 : Operatorcontent : GroupAggregate(groupBy=[name], select=[name, COUNT(*) AS cnt])ship_strategy : HASH

結果展示

SQL Client支持兩種模式來維護并展示查詢結果：

table mode

在內存中物化查詢結果，并以分頁table形式展示。用戶可以通過以下命令啟用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 Query Result (Table)Table program finished.                                                                                       Page: Last of 1                                                                                         Updated: 16:55:08.589name                       cntAlice                         1Greg                         1Bob                         2Q Quit                                         + Inc Refresh                                  G Goto Page                                    N Next Page                                    O Open Row
R Refresh                                      - Dec Refresh                                  L Last Page                                    P Prev Page

? [點擊并拖拽以移動] ??

changelog mode

不會物化查詢結果，而是直接對continuous query產生的添加和撤回retractions結果進行展示：如下案例中的-表示撤回消息

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 Query Result (Changelog)Table program finished.                                                                                                                                                                                               Updated: 16:58:05.777+/-                      name                       cnt+                       Bob                         1+                     Alice                         1+                      Greg                         1-                       Bob                         1+                       Bob                         2Q Quit                                                                        + Inc Refresh                                                                 O Open Row
R Refresh                                                                     - Dec Refresh

? [點擊并拖拽以移動] ??

Environment Files

CREATE TABLE 創建表DDL語句：

Flink SQL> CREATE TABLE pvuv_sink (
>     dt VARCHAR,
>     pv BIGINT,
>     uv BIGINT
> ) ;
[INFO] Table has been created.

SHOW TABLES 查看所有表名

Flink SQL>  show tables;
pvuv_sink

DESCRIBE 表名 查看表的詳細信息；

Flink SQL>  describe pvuv_sink;
root|-- dt: STRING|-- pv: BIGINT|-- uv: BIGINT

插入等操作均與關系型數據庫操作語句一樣，省略N個操作

Restful API

接下來我們演示如何通過rest api來提交jar包和執行任務。
[點擊并拖拽以移動] ?

通過Show Plan可以看到執行圖
[點擊并拖拽以移動] ?

提交之后的操作，取消的話點擊頁面的Cancel Job

? [點擊并拖拽以移動] ??