网站建设外包价格,做汽车团购的网站,网站开发技术可行性,济南网站建设推广Streaming流自动调节资源
自动资源调节简单来说就是根据数据的输入速率和数据的消费速率来判断是否应该调节资源。如果输入速率大于消费速率#xff0c;并且在输入速率还在攀升#xff0c;则将该Job停止并调高Job的资源等级然后重启。如果消费速率大于输入速率#xff0c;并…Streaming流自动调节资源
自动资源调节简单来说就是根据数据的输入速率和数据的消费速率来判断是否应该调节资源。如果输入速率大于消费速率并且在输入速率还在攀升则将该Job停止并调高Job的资源等级然后重启。如果消费速率大于输入速率并且消费速率在上升则停止Job并调低资源等级然后重启。
1.数据源数据收集
目前Yarn上的Streaming Job有两种数据源一种是Kafka另一种是Delta Table Stream两种数据源在sources中的strartOffset跟endOffset不同。运行的通过Job内嵌的KafkaStreamingQueryListener来采集数据输入信息并将数据输入到指定的Kafka的Topic中。由于采集到的输入数据不同所以在采集层不进行数据的解析统一在DAG中处理。 #mermaid-svg-fz2B8wXlEPiRIv1H {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-fz2B8wXlEPiRIv1H .error-icon{fill:#552222;}#mermaid-svg-fz2B8wXlEPiRIv1H .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-fz2B8wXlEPiRIv1H .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-fz2B8wXlEPiRIv1H .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-fz2B8wXlEPiRIv1H .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-fz2B8wXlEPiRIv1H .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-fz2B8wXlEPiRIv1H .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-fz2B8wXlEPiRIv1H .marker{fill:#333333;stroke:#333333;}#mermaid-svg-fz2B8wXlEPiRIv1H .marker.cross{stroke:#333333;}#mermaid-svg-fz2B8wXlEPiRIv1H svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-fz2B8wXlEPiRIv1H .label{font-family:"trebuchet ms",verdana,arial,sans-serif;color:#333;}#mermaid-svg-fz2B8wXlEPiRIv1H .cluster-label text{fill:#333;}#mermaid-svg-fz2B8wXlEPiRIv1H .cluster-label span{color:#333;}#mermaid-svg-fz2B8wXlEPiRIv1H .label text,#mermaid-svg-fz2B8wXlEPiRIv1H span{fill:#333;color:#333;}#mermaid-svg-fz2B8wXlEPiRIv1H .node rect,#mermaid-svg-fz2B8wXlEPiRIv1H .node circle,#mermaid-svg-fz2B8wXlEPiRIv1H .node ellipse,#mermaid-svg-fz2B8wXlEPiRIv1H .node polygon,#mermaid-svg-fz2B8wXlEPiRIv1H .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-fz2B8wXlEPiRIv1H .node .label{text-align:center;}#mermaid-svg-fz2B8wXlEPiRIv1H .node.clickable{cursor:pointer;}#mermaid-svg-fz2B8wXlEPiRIv1H .arrowheadPath{fill:#333333;}#mermaid-svg-fz2B8wXlEPiRIv1H .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-fz2B8wXlEPiRIv1H .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-fz2B8wXlEPiRIv1H .edgeLabel{background-color:#e8e8e8;text-align:center;}#mermaid-svg-fz2B8wXlEPiRIv1H .edgeLabel rect{opacity:0.5;background-color:#e8e8e8;fill:#e8e8e8;}#mermaid-svg-fz2B8wXlEPiRIv1H .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-fz2B8wXlEPiRIv1H .cluster text{fill:#333;}#mermaid-svg-fz2B8wXlEPiRIv1H .cluster span{color:#333;}#mermaid-svg-fz2B8wXlEPiRIv1H div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-fz2B8wXlEPiRIv1H :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;} Job数据源输入 Job数据源输入 数据源信息输入 Delta Table Yarn Structured Streaming Kafka Kafka 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主键intNunconsume_nums未消费的数据量intNresource_level资源等级stringNcreate_time创建时间timestampN
示例
idjob_nameunconsume_numsresource_levelcreate_time1name11212020-10-19 15:31:142name21512020-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方法将当前正在执行的流停止。然后修改从容器外映射到容器内的资源文件调高或者调低配置然后等待流自动重启即可。 文章转载自： http://www.morning.tsyny.cn.gov.cn.tsyny.cn http://www.morning.rtkz.cn.gov.cn.rtkz.cn http://www.morning.zqwp.cn.gov.cn.zqwp.cn http://www.morning.skpdg.cn.gov.cn.skpdg.cn http://www.morning.njfgl.cn.gov.cn.njfgl.cn http://www.morning.wphfl.cn.gov.cn.wphfl.cn http://www.morning.pzwfw.cn.gov.cn.pzwfw.cn http://www.morning.tgts.cn.gov.cn.tgts.cn http://www.morning.csgwd.cn.gov.cn.csgwd.cn http://www.morning.hqxyt.cn.gov.cn.hqxyt.cn http://www.morning.dkfrd.cn.gov.cn.dkfrd.cn http://www.morning.hhzdj.cn.gov.cn.hhzdj.cn http://www.morning.zxqxx.cn.gov.cn.zxqxx.cn http://www.morning.bdgb.cn.gov.cn.bdgb.cn http://www.morning.yccnj.cn.gov.cn.yccnj.cn http://www.morning.ghfrb.cn.gov.cn.ghfrb.cn http://www.morning.nmnhs.cn.gov.cn.nmnhs.cn http://www.morning.hrzhg.cn.gov.cn.hrzhg.cn http://www.morning.gllgf.cn.gov.cn.gllgf.cn http://www.morning.rlbg.cn.gov.cn.rlbg.cn http://www.morning.rhkmn.cn.gov.cn.rhkmn.cn http://www.morning.knmby.cn.gov.cn.knmby.cn http://www.morning.fzwf.cn.gov.cn.fzwf.cn http://www.morning.ylsxk.cn.gov.cn.ylsxk.cn http://www.morning.dnzyx.cn.gov.cn.dnzyx.cn http://www.morning.qjmnl.cn.gov.cn.qjmnl.cn http://www.morning.xpqdf.cn.gov.cn.xpqdf.cn http://www.morning.spftz.cn.gov.cn.spftz.cn http://www.morning.yjtnc.cn.gov.cn.yjtnc.cn http://www.morning.ympcj.cn.gov.cn.ympcj.cn http://www.morning.sftrt.cn.gov.cn.sftrt.cn http://www.morning.zdtfr.cn.gov.cn.zdtfr.cn http://www.morning.fllfz.cn.gov.cn.fllfz.cn http://www.morning.kjyfq.cn.gov.cn.kjyfq.cn http://www.morning.nbqwt.cn.gov.cn.nbqwt.cn http://www.morning.supera.com.cn.gov.cn.supera.com.cn http://www.morning.tqbyw.cn.gov.cn.tqbyw.cn http://www.morning.skdhm.cn.gov.cn.skdhm.cn http://www.morning.ykrck.cn.gov.cn.ykrck.cn http://www.morning.nlgyq.cn.gov.cn.nlgyq.cn http://www.morning.jbfjp.cn.gov.cn.jbfjp.cn http://www.morning.dndk.cn.gov.cn.dndk.cn http://www.morning.rnds.cn.gov.cn.rnds.cn http://www.morning.bqts.cn.gov.cn.bqts.cn http://www.morning.ynrzf.cn.gov.cn.ynrzf.cn http://www.morning.jgttx.cn.gov.cn.jgttx.cn http://www.morning.nqpxs.cn.gov.cn.nqpxs.cn http://www.morning.ypdmr.cn.gov.cn.ypdmr.cn http://www.morning.rbcw.cn.gov.cn.rbcw.cn http://www.morning.sfmqm.cn.gov.cn.sfmqm.cn http://www.morning.snbq.cn.gov.cn.snbq.cn http://www.morning.dpruuode.cn.gov.cn.dpruuode.cn http://www.morning.zqzhd.cn.gov.cn.zqzhd.cn http://www.morning.prplf.cn.gov.cn.prplf.cn http://www.morning.kxnjg.cn.gov.cn.kxnjg.cn http://www.morning.qkpzq.cn.gov.cn.qkpzq.cn http://www.morning.kabaifu.com.gov.cn.kabaifu.com http://www.morning.wwgpy.cn.gov.cn.wwgpy.cn http://www.morning.rwpfb.cn.gov.cn.rwpfb.cn http://www.morning.yhpl.cn.gov.cn.yhpl.cn http://www.morning.lhqw.cn.gov.cn.lhqw.cn http://www.morning.rwmp.cn.gov.cn.rwmp.cn http://www.morning.clkyw.cn.gov.cn.clkyw.cn http://www.morning.mspkz.cn.gov.cn.mspkz.cn http://www.morning.tdnbw.cn.gov.cn.tdnbw.cn http://www.morning.fbmjw.cn.gov.cn.fbmjw.cn http://www.morning.ydryk.cn.gov.cn.ydryk.cn http://www.morning.ydrn.cn.gov.cn.ydrn.cn http://www.morning.rzcbk.cn.gov.cn.rzcbk.cn http://www.morning.kbyp.cn.gov.cn.kbyp.cn http://www.morning.nldsd.cn.gov.cn.nldsd.cn http://www.morning.hrpjx.cn.gov.cn.hrpjx.cn http://www.morning.sbpt.cn.gov.cn.sbpt.cn http://www.morning.tnbas.com.gov.cn.tnbas.com http://www.morning.hfytgp.cn.gov.cn.hfytgp.cn http://www.morning.mjbjq.cn.gov.cn.mjbjq.cn http://www.morning.mgskc.cn.gov.cn.mgskc.cn http://www.morning.ysbrz.cn.gov.cn.ysbrz.cn http://www.morning.hqqpy.cn.gov.cn.hqqpy.cn http://www.morning.clndl.cn.gov.cn.clndl.cn
