达州达县网站建设清博大数据舆情监测平台

Streaming流自动调节资源

自动资源调节简单来说就是根据数据的输入速率和数据的消费速率来判断是否应该调节资源。如果输入速率大于消费速率，并且在输入速率还在攀升，则将该Job停止并调高Job的资源等级然后重启。如果消费速率大于输入速率，并且消费速率在上升，则停止Job并调低资源等级然后重启。

1.数据源数据收集

目前Yarn上的Streaming Job有两种数据源，一种是Kafka另一种是Delta Table Stream，两种数据源在sources中的strartOffset跟endOffset不同。运行的通过Job内嵌的KafkaStreamingQueryListener来采集数据输入信息，并将数据输入到指定的Kafka的Topic中。由于采集到的输入数据不同，所以在采集层不进行数据的解析，统一在DAG中处理。

2.数据源数据处理

DAG运行之后，会先进行数据源数据的处理，从Variables中获取到上次消费的offset位置，从kafka中获取当前最后数据的offset，然后消费两个offset之前的数据，数据处理完之后将Variables的offset位置置为当前最后数据的offset。

​ 2.1. Delta Stream处理：从1中获取到的数据源数据Delta Table没有具体的offset信息，所以将reservoirVersion作为消费的offset标识，然后将相同数据源的数据的消费标识组合成source_info，根据数据源中JobName跟source_info的Delta Table地址，用DeltaTable的history方法获取Delta表中这个版本之后的版本numOutputRows之和作为未消费的数据量unconsume_nums。

​ 2.2. Kafka数据处理：从1中获取到的数据源Kafka数据能够获取到具体的offset信息，将startOffset作为消费标识，然后将相同数据源的数据的消费标识组合成source_info，把startOffset和endOffset中相同分区的offset做差再求和得到未消费的数据量unconsume_nums。

然后将处理完之后的数据持久化到Mysql中，表结构如下：

表名	streaming_consume_status
显示名称	实时流消费情况表
描述	记录流消费情况信息
Schema
粒度	每条记录表示每五分钟一个Job的消费情况
更新周期	每5分钟

表字段：

列名	描述	数据类型	空值
id	主键	int	N
unconsume_nums	未消费的数据量	int	N
resource_level	资源等级	string	N
create_time	创建时间	timestamp	N

示例：

id	job_name	unconsume_nums	resource_level	create_time
1	name1	12	1	2020-10-19 15:31:14
2	name2	15	1	2020-10-19 15:31:14

CREATE TABLE `streaming_consume_status` (`id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键ID',`job_name` varchar(64) COMMENT 'Job名称',`source_type` varchar(64) COMMENT '数据源类型',`unconsume_nums` int(11) COMMENT '未消费的数据数量',`resource_level` int(5) COMMENT '资源等级',`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',PRIMARY KEY (`id`)
);CREATE TABLE `streaming_query_offset` (`job_name` varchar(64) COMMENT 'Job名称',`query_name` varchar(64) COMMENT 'QUERY名称',`delta_path` varchar(500) COMMENT 'delta路径',`offset` int(11) COMMENT 'query当前消费的offset/version',`create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',`update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',PRIMARY KEY (`job_name`, `query_name`, `delta_path`)
);insert into streaming_query_offset (job_name, query_name, offset) values('product2.ods.streamingLogSinkJob', 'jclog_delta', 1000)INSERT INTO monitor.streaming_query_offset (job_name, query_name, delta_path, offset) values ("stage2.cdm.dim.streamingDimSCDJob", "gggg", 'cccc', 888) ON DUPLICATE KEY UPDATE offset = 888;INSERT INTO streaming_query_offset (job_name, query_name, offset) VALUES ('product2.ods.streamingLogSinkJob', 'elog_delta', 1003) ON DUPLICATE KEY UPDATE offset = 1003; 

3.资源调整策略

调整策略是根据一段时间内，输入数据量跟消费数据量的差值计算出的斜率曲线来判断。比如说取近30分钟的输入输出数据，从数据库会获得6条数据，根据消费数据量consumer_nums和生产数据量producer_nums计算出消费差值，看这个周期内的差值变化。

调高等级：如果计算出来的5个斜率都大于设置的阈值，就认为消费速率比生产速率慢很多，需要调高资源等级并重启流任务。

调低等级：如果计算出来的5个斜率都小于设置的阈值，就认为消费速率比生产速率快很多，需要调低资源等级并重启流任务来节省资源。

特殊情况，如果斜率都为0，会查看当前资源等级，如果不是最低则逐次降低资源等级。

如果斜率在-1到1之前，则认为消费速率正常，不需要调整资源等级。

DAG在运行到这步时，会从Mysql中读job_name为XX，并且时间最晚的6条数据，通过这六条数据来计算出斜率曲线。

4.资源调整

DAG在上一步判断出是否要调整资源之后，会向企业微信发报警通知当前Job发生资源调整操作，然后根据对应的JobName通过yarn client api获取当前的ApplicationId，然后调用api的kill方法将当前正在执行的流停止。然后修改从容器外映射到容器内的资源文件，调高或者调低配置，然后等待流自动重启即可。