1.2

master:9092,node1:9092,node2:9092,

指定kafka集群列表

在内存中实体化结果并将结果用规则的分页表格可视化展示出来。

执行如下命令启用

SET

sql-client.execution.result-mode

table;

不会实体化和可视化结果而是由插入和撤销-组成的持续查询产生结果流。

开启命令

SET

sql-client.execution.result-mode

changelog;

我在kafka生产端添加了一条理科六班的数据他表的变化是先加后减然后再加他就是属于update有之前的数据更新

我又在kafka生产端添加了一条理科六班1的数据这张动态表之前没有这个数据所以他是insert

2.2.3

1.更接近传统的数据库会将执行的结果以制表的形式直接打在屏幕之上。

具体显示的内容会取决于作业

2.命令

sql-client.execution.result-mode

tableau;

类似于日志一样不会新开一个窗口数据的添加还是跟变更日志模式一样

但是添加已有数据的方法是不一样的他是先减后加他是Retract流添加之前没有数据还是insert

2.3

dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka/artifactIdversion1.15.4/version

/dependency

master:9092,node1:9092,node2:9092,

properties.group.id

id,name,event_time,partition,offset

from

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092

--from-beginning

{data:[{clazz:理科六班,num:377}],type:INSERT}

CREATE

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092

--from-beginning

flink-connector-jdbc-1.15.2.jar

mysql-connector-java-5.1.47.jar

依赖更新后需要重启集群才会生效

jdbc:mysql://master:3306/bigdata29,

table-name

1.将flink-sql中的一张表或者查询语句的结果存放在mysql中

2.mysql中表需要提前创建

jdbc:mysql://master:3306/bigdata29?useUnicodetruecharacterEncodingUTF-8,

table-name

hdfs://master:9000/bigdata29/data/students.csv,

必选指定路径

hdfs://master:9000/data/students,

必选指定路径

hdfs://master:9000/data/students_sink,

必选指定路径

hdfs://master:9000/data/clazz_num,

必选指定路径

flink-sql-connector-hbase-2.2-1.15.2.jar

依赖更新后需要重启集群才会生效

master:2181,node1:2181,node2:2181

5.查看结果

master:9092,node1:9092,node2:9092,

指定kafka集群列表

1.flink表中的字段和类型需要和json中保持一致同名映射

2.建表

master:9092,node1:9092,node2:9092,

指定kafka集群列表

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092,

properties.group.id

OPTIONS(csv.ignore-parse-errors

true)

master:9092,node1:9092,node2:9092,

指定kafka集群列表

1.当有一段sql逻辑重复时可以定义在with语句中减少代码量

2.sql

2、如果表的数据量很大随着时间的推移状态会越来越大状态的数据时先保存在TM的内存中的时间长了可能会出问题

注意distinct后面加字段

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092

--topic

item,price,bidtime,window_start,window_end,window_time

FROM

其中TUMBLE:滚动窗口函数在原表的基础上增加窗口开始时间窗口结束时间窗口时间重新组成一张表

DESCRIPTOR(bidtime)里面传入的是事件时间INTERVAL

2.滑动窗口函数

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092

--topic

item,price,proctime,window_start,window_end,window_time

FROM

HOP滑动窗口函数在原表的基础上增加窗口开始时间窗口结束时间窗口时间重新组成一张表

INTERVAL

SESSION_START(proctime,INTERVAL

SECOND)

如果状态无限增长会导致checkpoint时间拉长如果checkpoint超时失败了也会导致任务失败

insert

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092

--topic

2、实时统计每个商品的累计总金额将总金额放在每一条数据的后面只统计最近10分钟的数据

select

2、实时统计每个商品的累计总金额将总金额放在每一条数据的后面计算最近5条数据

select

2、实时统计每个商品的最大金额将总金额放在每一条数据的后面计算最近5条数据

select

2.本来开窗的字段只能是时间字段如果外面接一个子查询那么就没有限制了

TOPN

1.考虑计算代价只能按照时间字段来进行升序(第一个字段必须是时间字段后面可以是非时间字段)

select

不考虑计算代价加上子查询或者limit限制那么可以是非时间字段

select

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092

--topic

master:9092,node1:9092,node2:9092,

properties.group.id

master:9092,node1:9092,node2:9092

--topic

常规的关联方式会将两个表的数据一直保存在状态中时间长了状态会越来越大导致任务执行失败

状态有效期状态在flink中保存的事件状态保留多久需要根据实际业务分析

SET

1.在一段时间内关联字段必须要有时间戳的时间字段两张表都是流式的。

select

a表的时间需要在b表时间10秒内或者b表的时间需要在a表时间10秒内

and

master:9092,node1:9092,node2:9092,

指定kafka集群列表

master:9092,node1:9092,node2:9092

--topic

master:9092,node1:9092,node2:9092,

指定kafka集群列表

master:9092,node1:9092,node2:9092

--from-beginning

a.price,a.order_time,b.conversion_rate,b.update_time

from

a.price,a.order_time,b.conversion_rate,b.update_time

from

维度表不怎么变化的变维度表的数据一般可以放在hdfs或者mysql

2.建表语句

master:9092,node1:9092,node2:9092,

指定kafka集群列表

master:9092,node1:9092,node2:9092

--topic

jdbc:mysql://master:3306/bigdata29,

table-name

如果mysql表里面的数据增加的话刚刚好增加的部分数据能与流表数据关联但是查询不到

4.lookup

flink-sql-connector-hive-3.1.2_2.12-1.15.2.jar

重启flink集群

1.catalog---database---table----字段----数据

catalog是数据库上面的一个概念一个cataloglog中可以有多个database,

2.default_catalog是flink默认的元数据将元数据保存在jobmanager的内存中

3.使用

/usr/local/soft/hive-3.1.2/conf

show

hive_catalog.bigdata29.students;

创建数据库

flink可以查询hive的表hive不能查询flink创建的动态表

在hive

master:9092,node1:9092,node2:9092,

指定kafka集群列表

master:9092,node1:9092,node2:9092,

指定kafka集群列表

master:9092,node1:9092,node2:9092

--topic

master:9092,node1:9092,node2:9092,

指定kafka集群列表

hdfs://master:9000/flink/checkpoint/d915e6278f156a9278156e67105f914e/chk-36;

3.重新提交

master:9092,node1:9092,node2:9092,

指定kafka集群列表

执行一组sql,如果多个sql中使用了同一张表flink只会读取一次

EXECUTE

OPTIONS(fields.word.length5,rows-per-second400000)

group

table.exec.mini-batch.allow-latency

set

table.optimizer.agg-phase-strategy

TWO_PHASE;