Shell、SQL

在Flink安装下可以看到flink#xff0c;start-scala-shell.sh和sql-client.sh等文件#xff0c;这些都是客户…Flink提供了丰富的客户端操作来提交任务和与任务进行交互。

下面主要从Flink命令行、Scala

Shell、SQL

在Flink安装下可以看到flinkstart-scala-shell.sh和sql-client.sh等文件这些都是客户端操作的入口。

flink

-c如果没有在jar包中指定入口类则需要在这里通过这个参数指定

-m指定需要连接的jobmanager(主节点)地址使用这个参数可以指定一个不同于配置文件中的jobmanager可以说是yarn集群名称

-s保存点savepoint的路径以还原作业来自例如hdfs:///flink/savepoint-1537)

[roothadoop1

examples/streaming/TopSpeedWindowing.jar

Executing

dce7b69ad15e8756766967c46122736f就可以看到我们提交的JobManager默认是一个并发。

点击左侧TaskManager

或者查看TaskManager节点的log目录下的*.out文件也能看到具体的输出信息。

list

dce7b69ad15e8756766967c46122736f

(RUNNING)

--------------------------------------------------------------

scheduled

需要指定jobmanager的ip:prot和jobId。

如下报错可知一个job能够被stop要求所有的source都是可以stoppable的即实现了

[roothadoop1

dce7b69ad15e8756766967c46122736f

Suspending

dce7b69ad15e8756766967c46122736f

with

savepoint.------------------------------------------------------------The

program

exception:org.apache.flink.util.FlinkException:

Could

dce7b69ad15e8756766967c46122736f.at

org.apache.flink.client.cli.CliFrontend.lambda$stop$5(CliFrontend.java:458)StoppableFunction接口如下属于优雅停止任务。

/***

如果在conf/flink-conf.yaml里面配置state.savepoints.dir会保存savepoint否则不会保存savepoint。

重启

Cancel命令

e8ce0d111262c52bf8228d5722742d47

DEPRECATION

e8ce0d111262c52bf8228d5722742d47

with

e8ce0d111262c52bf8228d5722742d47.

Savepoint

file:/tmp/savepoint/savepoint-e8ce0d-f7fa96a085d8.也可以在停止的时候显示指定savepoint目录

[roothadoop1

f58bb4c49ee5580ab5f27fdb24083353

DEPRECATION

f58bb4c49ee5580ab5f27fdb24083353

with

f58bb4c49ee5580ab5f27fdb24083353.

Savepoint

file:/tmp/savepoint/savepoint-f58bb4-127b7e84910e.取消和停止流作业的区别如下

cancel()调用

立即调用作业算子的cancel()方法以尽快取消它们。

如果算子在接到cancel()调用后没有停止Flink将开始定期中断算子线程的执行直到所有算子停止为止。

stop()调用

是更优雅的停止正在运行流作业的方式。

stop()仅适用于source实现了StoppableFunction接口的作业。

当用户请求停止作业时作业的所有source都将接收stop()方法调用。

直到所有source正常关闭时作业才会正常结束。

这种方式使

savepoint

当需要生成savepoint文件时需要手动触发savepoint。

如下需要指定正在运行的

JobID

和生成文件的存放目录。

同时我们也可以看到它会返回给用户存放的savepoint的文件名称等信息。

[roothadoop1

examples/streaming/TopSpeedWindowing.jar

Executing

216c427d63e3754eb757d2cc268a448d[roothadoop1

flink-1.10.1]#

216c427d63e3754eb757d2cc268a448d

savepoint

216c427d63e3754eb757d2cc268a448d.Waiting

for

file:/tmp/savepoint/savepoint-216c42-154a34cf6bfdYou

can

command.savepoint和checkpoint的区别

checkpoint是增量做的每次的时间较短数据量较小只要在程序里面启用后会自动触发用户无须感知savepoint是全量做的每次的时间较长数据量较大需要用户主动去触发。

checkpoint是作业failover的时候自动使用不需要用户指定。

savepoint一般用于程序的版本更新bug修复A/B

从指定

/tmp/savepoint/savepoint-f58bb4-127b7e84910e/

examples/streaming/TopSpeedWindowing.jar

Executing

1a5c5ce279e0e4bd8609f541b37652e2查看JobManager的日志能够看到Reset

the

这里修改master的conf/flink-conf.yaml将task

[roothadoop1

examples/streaming/TopSpeedWindowing.jar

Executing

2e833a438da7d8052f14d5433910515a从页面上能看到Task

Slots总计变为了8运行的Slot为1剩余Slot数量为7。

Flink1.0版本命令行flink

modify已经没有这个行为了被移除了。

。

。

Flink1.7上是可以运行的。

[roothadoop1

cc22cc3d09f5d65651d637be6fb0a1c3

modify

examples/streaming/TopSpeedWindowing.jar

Execution

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,

PassThroughWindowFunction),pact:Operator,contents:Window(GlobalWindows(),

DeltaTrigger,

PassThroughWindowFunction),parallelism:1,predecessors:[{id:2,ship_strategy:HASH,side:second}]},{id:5,type:Sink:

Print

Out,parallelism:1,predecessors:[{id:4,ship_strategy:FORWARD,side:second}]}]}

--------------------------------------------------------------拷贝输出的json内容粘贴到这个网站http://flink.apache.org/visualizer/可以生成类似如下的执行图。

SQL

Page打开http://hadoop1:8081能看到这条select语句产生的查询任务已经结束了。

这个查询采用的是读取固定数据集的Custom

Collect

LogicalValues(type[RecordType(VARCHAR(5)

name)],

Exchange(distribution[hash[name]])-

Values(type[RecordType(VARCHAR(5)

name)],

在内存中物化查询结果并以分页table形式展示。

用户可以通过以下命令启用table

mode例如如下案例

query产生的添加和撤回retractions结果进行展示如下案例中的-表示撤回消息

Flink

execution.result-modechangelog;

[INFO]

BIGINT插入等操作均与关系型数据库操作语句一样省略N个操作

Restful