Flink 支持使用 HiveCatalog 来管理 Flink 的元数据信息, HiveCatalog 通过授权用户仅创建一次表和其他元数据对象，这样就避免了重复创建 kafka 流表,节省了大量的时间, 对于不同的用户来说,可以直接使用表而不需要再去创建.
就拿消费 kafka 来说,经常会有下面的需求:
•用户需要指定特性的消费时间戳，即修改 scan.startup.timestamp-millis 属性.•用户想忽略掉解析错误，需要将 format.ignore-parse-errors 改为 true.•不同用户想要设置不同的 group.id 去消费数据.•用户想要改变消费 kafka 的策略比如从最早的 offset 开始消费数据.
类似于这样的需求很常见,有的时候只是临时修改一下属性,难道需要把表删了,再重新创建新的表吗 ? 显然不是的,下面就介绍两种方式去修改表的属性.
1, ALTER TABLE
•重命名表
ALTER TABLE [catalog_name.][db_name.]table_name RENAME TO new_table_name
将给定的表名重命名为另一个新的表名。
•设置或更改表属性
ALTER TABLE [catalog_name.][db_name.]table_name SET (key1=val1, key2=val2, ...)
在指定的表中设置一个或多个属性。如果表中已经设置了特定属性，会用新属性覆盖旧值。
2, Dynamic Table Options
从 1.11 开始，用户可以通过动态参数的形式灵活地设置表的属性参数，覆盖或者追加原表的 WITH (...) 语句内定义的 table options。
基本语法为：
table_path /*+ OPTIONS(key=val [, key=val]*) */
动态参数的使用没有语境限制，只要是引用表的地方都可以追加定义。在指定的表后面追加的动态参数会自动追加到原表定义中，是不是很方便呢
由于可能对查询结果有影响，动态参数功能默认是关闭的， 使用下面的方式开启该功能：
set table.dynamic-table-options.enabled=true;
下面就来分别演示一下这两种方法的具体操作
3, 建表
-
DROP
TABLE
IF
EXISTS KafkaTable;
-
CREATE
TABLE KafkaTable (
-
`age`
BIGINT,
-
`name`
STRING,
-
`ts`
TIMESTAMP(
3) METADATA
FROM
'timestamp',
-
`partition`
BIGINT METADATA
VIRTUAL,
-
`offset`
BIGINT METADATA
VIRTUAL,
-
`topic`
STRING METADATA
VIRTUAL,
-
WATERMARK
FOR ts
AS ts -
INTERVAL
'0'
SECOND
-
)
WITH (
-
'connector' =
'kafka',
-
'topic' =
'new_flink_topic',
-
'properties.bootstrap.servers' =
'master:9092,storm1:9092,storm2:9092',
-
'properties.group.id' =
'flink_jason',
-
'scan.startup.mode' =
'latest-offset',
-
'format' =
'json',
-
'json.fail-on-missing-field' =
'false',
-- 字段丢失任务不失败
-
'json.ignore-parse-errors' =
'true'
-- 解析失败跳过
-
);
-
DROP
TABLE
IF
EXISTS print_table;
-
CREATE
TABLE print_table
-
(
-
name
string,
-
pv
BIGINT,
-
uv
BIGINT
-
)
-
WITH (
'connector' =
'print');
-
insert
into print_table
-
SELECT
name,
-
count(*)
as pv,
-
count(
distinct
name)
as uv
-
from KafkaTable
-
group
by
name;
上面把 3 条语句放在一块执行了,然后先在 Flink SQL 里面查一下刚才建的 kafka 流表和打印数据的结果表
然后再到 hive 里面查一下这两个表是否存在
可以看到这两个表在 hive 里面也是存在的,因为已经配置了 HiveCatalog 后面就不用再重复建表了,可以直接使用这两个表.
再来看一下 hive 里面的表结构信息.
-
hive>
show
create
table kafkatable ;
-
OK
-
CREATE
TABLE
`kafkatable`(
-
)
-
ROW
FORMAT SERDE
-
'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
-
STORED
AS INPUTFORMAT
-
'org.apache.hadoop.mapred.TextInputFormat'
-
OUTPUTFORMAT
-
'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'
-
LOCATION
-
'hdfs://master:9000/hive/warehouse/mydatabase.db/kafkatable'
-
TBLPROPERTIES (
-
'flink.connector'=
'kafka',
-
'flink.format'=
'json',
-
'flink.json.fail-on-missing-field'=
'false',
-
'flink.json.ignore-parse-errors'=
'true',
-
'flink.properties.bootstrap.servers'=
'master:9092,storm1:9092,storm2:9092',
-
'flink.properties.group.id'=
'flink_jason',
-
'flink.scan.startup.mode'=
'latest-offset',
-
'flink.schema.0.data-type'=
'BIGINT',
-
'flink.schema.0.name'=
'age',
-
'flink.schema.1.data-type'=
'VARCHAR(2147483647)',
-
'flink.schema.1.name'=
'name',
-
'flink.schema.2.data-type'=
'TIMESTAMP(3)',
-
'flink.schema.2.metadata'=
'timestamp',
-
'flink.schema.2.name'=
'ts',
-
'flink.schema.2.virtual'=
'false',
-
'flink.schema.3.data-type'=
'BIGINT',
-
'flink.schema.3.metadata'=
'partition',
-
'flink.schema.3.name'=
'partition',
-
'flink.schema.3.virtual'=
'true',
-
'flink.schema.4.data-type'=
'BIGINT',
-
'flink.schema.4.metadata'=
'offset',
-
'flink.schema.4.name'=
'offset',
-
'flink.schema.4.virtual'=
'true',
-
'flink.schema.5.data-type'=
'VARCHAR(2147483647)',
-
'flink.schema.5.metadata'=
'topic',
-
'flink.schema.5.name'=
'topic',
-
'flink.schema.5.virtual'=
'true',
-
'flink.schema.watermark.0.rowtime'=
'ts',
-
'flink.schema.watermark.0.strategy.data-type'=
'TIMESTAMP(3)',
-
'flink.schema.watermark.0.strategy.expr'=
'`ts` - INTERVAL \'
0\
' SECOND',
-
'flink.topic'=
'new_flink_topic',
-
'is_generic'=
'true',
-
'transient_lastDdlTime'=
'1608409788')
-
Time taken:
0.046
seconds, Fetched:
44
row(s)
可以看到 'flink.scan.startup.mode'='latest-offset' 这个属性是从最新的位置开始消费数据的.然后我们来看一下 Flink 的 UI
因为是从最新的 offset 位置开始消费的,我没有写入新数据,所以消费到的数据是 0 条.然后我把任务停掉, 加上动态表选项,修改为从头开始消费,然后再提交任务.
-
insert
into print_table
-
SELECT
name,
-
count(*)
as pv,
-
count(
distinct
name)
as uv
-
from KafkaTable
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */
-
group
by
name;
可以看到在我没有写入新数据的情况下,已经消费到 534 条数据.也可以到 tm 的 stdout 里面观察一下数据打印的变化情况,说明刚才修改的 'scan.startup.mode'='earliest-offset' 属性是起作用的,任务是从头开始消费数据.
然后我再用 alter table 来直接修改表里面的属性.
alter table myhive.mydatabase.KafkaTable set('scan.startup.mode'='earliest-offset');
先不提交任务,先到 hive 表里看下属性是否修改了.
-
>
show
create
table kafkatable ;
-
OK
-
CREATE
TABLE
`kafkatable`(
-
)
-
ROW
FORMAT SERDE
-
'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
-
STORED
AS INPUTFORMAT
-
'org.apache.hadoop.mapred.TextInputFormat'
-
OUTPUTFORMAT
-
'org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat'
-
LOCATION
-
'hdfs://master:9000/hive/warehouse/mydatabase.db/kafkatable'
-
TBLPROPERTIES (
-
'flink.connector'=
'kafka',
-
'flink.format'=
'json',
-
'flink.json.fail-on-missing-field'=
'false',
-
'flink.json.ignore-parse-errors'=
'true',
-
'flink.properties.bootstrap.servers'=
'master:9092,storm1:9092,storm2:9092',
-
'flink.properties.group.id'=
'flink_jason',
-
'flink.scan.startup.mode'=
'earliest-offset',
-
'flink.schema.0.data-type'=
'BIGINT',
-
'flink.schema.0.name'=
'age',
-
'flink.schema.1.data-type'=
'VARCHAR(2147483647)',
-
'flink.schema.1.name'=
'name',
-
'flink.schema.2.data-type'=
'TIMESTAMP(3)',
-
'flink.schema.2.metadata'=
'timestamp',
-
'flink.schema.2.name'=
'ts',
-
'flink.schema.2.virtual'=
'false',
-
'flink.schema.3.data-type'=
'BIGINT',
-
'flink.schema.3.metadata'=
'partition',
-
'flink.schema.3.name'=
'partition',
-
'flink.schema.3.virtual'=
'true',
-
'flink.schema.4.data-type'=
'BIGINT',
-
'flink.schema.4.metadata'=
'offset',
-
'flink.schema.4.name'=
'offset',
-
'flink.schema.4.virtual'=
'true',
-
'flink.schema.5.data-type'=
'VARCHAR(2147483647)',
-
'flink.schema.5.metadata'=
'topic',
-
'flink.schema.5.name'=
'topic',
-
'flink.schema.5.virtual'=
'true',
-
'flink.schema.watermark.0.rowtime'=
'ts',
-
'flink.schema.watermark.0.strategy.data-type'=
'TIMESTAMP(3)',
-
'flink.schema.watermark.0.strategy.expr'=
'`ts` - INTERVAL \'
0\
' SECOND',
-
'flink.topic'=
'new_flink_topic',
-
'is_generic'=
'true',
-
'transient_lastDdlTime'=
'1608411139')
-
Time taken:
0.043
seconds, Fetched:
44
row(s)
'flink.scan.startup.mode'='earliest-offset' 可以看到这个属性已经被修改了.然后直接提交任务就可以了,不需要再加刚才的那个动态表属性.
还是刚才的 534 条数据,这两种方式都可以解决上面的用户需求.
区别:
alter table 的方式是直接修改表的属性, 元数据信息也会被修改, Dynamic Table Options 的方式只是通过动态参数的形式追加或者覆盖原表的属性,但是不会修改元数据信息.建议在生产环境还是使用第二种方式比较好.
转载：https://blog.csdn.net/xianpanjia4616/article/details/111465875