sparkConf常见参数设置

def getSparkConf():SparkConf = {val sparkConf: SparkConf = new SparkConf().set("spark.driver.cores","4")  //设置driver的CPU核数.set("spark.driver.maxResultSize","2g") //设置driver端结果存放的最大容量,这里设置成为2G,超过2G的数据,job就直接放弃,不运行了.set("spark.driver.memory","4g")  //driver给的内存大小.set("spark.executor.memory","8g")// 每个executor的内存.set("spark.submit.deployMode","cluster")  //spark 任务提交模式,线上使用cluster模式,开发使用client模式.set("spark.worker.timeout" ,"500") //基于standAlone模式下提交任务,worker的连接超时时间.set("spark.cores.max" , "10")  //基于standAlone和mesos模式下部署,最大的CPU和数量.set("spark.rpc.askTimeout" , "600s")  //spark任务通过rpc拉取数据的超时时间.set("spark.locality.wait" , "5s") //每个task获取本地数据的等待时间,默认3s钟,如果没获取到,依次获取本进程,本机,本机架数据.set("spark.task.maxFailures" , "5")  //允许最大失败任务数,根据自身容错情况来定.set("spark.serializer" ,"org.apache.spark.serializer.KryoSerializer")  //配置序列化方式.set("spark.streaming.kafka.maxRatePerPartition" , "5000")  //使用directStream方式消费kafka当中的数据,获取每个分区数据最大速率.set("spark.streaming.backpressure.enabled" , "true")  //开启sparkStreaming背压机制,接收数据的速度与消费数据的速度实现平衡//  .set("spark.streaming.backpressure.pid.minRate","10").set("spark.driver.host", "localhost")  //配置driver地址//shuffle相关参数调优开始.set("spark.reducer.maxSizeInFlight","96m")  //reduceTask拉取map端输出的最大数据量,调整太大有OOM的风险.set("spark.shuffle.compress","true")  //开启shuffle数据压缩.set("spark.default.parallelism","10")  //设置任务的并行度.set("spark.files.fetchTimeout","120s")  //设置文件获取的超时时间//网络相关参数.set("spark.rpc.message.maxSize","256")  //RPC拉取数据的最大数据量,单位M.set("spark.network.timeout","120s")  //网络超时时间设置.set("spark.scheduler.mode","FAIR")  //spark 任务调度模式  使用 fair公平调度//spark任务资源动态划分  https://spark.apache.org/docs/2.3.0/job-scheduling.html#configuration-and-setup.set("spark.dynamicAllocation.enabled","true").set("spark.shuffle.service.enabled","true").set("spark.dynamicAllocation.executorIdleTimeout","120s")  //executor空闲时间超过这个值,该executor就会被回收.set("spark.dynamicAllocation.minExecutors","0")  //最少的executor个数.set("spark.dynamicAllocation.maxExecutors","32")  //最大的executor个数  根据自己实际情况调整.set("spark.dynamicAllocation.initialExecutors","4")//初始executor个数.set("spark.dynamicAllocation.schedulerBacklogTimeout","5s")  //pending 状态的task时间,过了这个时间继续pending ,申请新的executor.setMaster("local[1]").setAppName("Stream")sparkConf.set("spark.speculation", "true")   //开启推测执行sparkConf.set("spark.speculation.interval", "100s")  // 每隔多久检测一次是否需要进行推测执行任务sparkConf.set("spark.speculation.quantile","0.9")  //完成任务的百分比,然后才能启动推测执行sparkConf.set("spark.streaming.backpressure.initialRate" , "500")  // //开启sparkStreaming的背压机制,然后第一批次获取数据的最大速率sparkConf.registerKryoClasses(Array(classOf[OrderInfo],classOf[Opt_alliance_business],classOf[DriverInfo],classOf[RegisterUsers]))sparkConf}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/508766.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

sparkSession常见参数设置

def getSparkSession(sparkConf:SparkConf):SparkSession {val sparkSession: SparkSession SparkSession.builder().config(sparkConf)//调度模式.config("spark.scheduler.mode", "FAIR").config("spark.executor.memoryOverhead", "51…

关于kafka中acks是否可以为all

kafka源码中有这样一段代码: org.apache.kafka.clients.producer.KafkaProducer private static int parseAcks(String acksString) {try {return acksString.trim().equalsIgnoreCase("all") ? -1 : Integer.parseInt(acksString.trim());} catch (Numb…

关于统计时间切片标签的一些sql

------当天付费明细表 DROP TABLE IF EXISTS rpt.tmp_mm_rb_daily_ffmx; create table rpt.tmp_mm_rb_daily_ffmx as select a.* FROM (select c.feemsisdn, c.destmsisdn, c.day, c.price/1000 fee, c.contentid, dc.content_name, c.ordernumber, c.cdrtime, c.createtime, c…

hadoop 二次开发DatanodeWriteTimeout设置

int getDatanodeWriteTimeout(int numNodes) {return this.dfsClientConf.confTime > 0 ? this.dfsClientConf.confTime 5000 * numNodes : 0;}int getDatanodeReadTimeout(int numNodes) {return this.dfsClientConf.socketTimeout > 0 ? 5000 * numNodes this.dfsC…

聚类算法

假定样本集 D {X1, 的,…, Xm} 包含 m 个无标记样本, 每个样本 X (X1; X2;… ; Xn) 是一个 n 维特征向量,则聚类算法将样本 集 D 划分为 k 个不相交的簇 {Gl I l 1, 2;… ,时,其中…

k-means均值向量

给定样本集 D {Xl) 的,… ,xm}, “k 均值” (k-means )算法针对聚类所 得簇划分 C {C1, C2,…, Ck} 最小化平方误差 ι ELL Ilx 一队IIL il EGi 其中队甘il LEGi X 是簇 q 的均值向量.在一定程度上 刻画了簇内样本围绕簇均值向量的紧密程度, E 值越小则…

学习向量量化

与 k 均值算法类似,“学习向量量化” (Learning Vector Quantization,简 称 LVQ)也是试图找到一组原型向量来刻画聚类结构, 但与一般聚类算法不同 的是, LVQ 假设数据样本带有类别标记,学习过程利用样本的这些监督信息…

k 近邻加权平均

k 近邻(k-Nearest Neighbor,简称 kNN)学习是一种常用的监督学习方法, 其工作机制非常简单: 给定测试样本?基于某种距离度量找出训练集中与其最 靠近的 k 个训练样本,然后基于这 k 个"邻居"的信息来进行预测. 通常, 在分…

k 近邻降维

k 近邻(k-Nearest Neighbor,简称 kNN)学习是一种常用的监督学习方法, 其工作机制非常简单: 给定测试样本?基于某种距离度量找出训练集中与其最 靠近的 k 个训练样本,然后基于这 k 个"邻居"的信息来进行预测. 通常, 在分…

维度建模工具

幵始维度建模工作前,项目组需要理解业务需求,以及作为基础的源数据的实际情况。 通过与、 Ik务代表交流来发现需求,用于理解他们的基于关键性能指标、竞争性商业问题、 决策制定过程、支持分析需求的目标。同时,数据实际情况可以通…

Cube和Grouping 和Rollup

增强的聚合 Cube和Grouping 和Rollup 这几个分析函数通常用于OLAP中,不能累加,而且需要根据不同维度上钻和下钻的指标统计,比如,分小时、天、月的UV数。 GROUPING SETS 在一个GROUP BY查询中,根据不同的维度组合进行聚…

常见维度建模错误

需要避免的常见维度建模错误 错误 10: 在事实表中放入文本属性 要从数据仓库事实表中 挑出这些文本属性,并将它们放入维度表中。 错误 9: 限制使用冗长的描述符以节省空间 维度表从几何上看总是比事实表小很多。 错误 8: 将层次划分为多个维度 以用户看来最自然最 有…

2020-09-21

columns has 234 elements while hbase.columns.mapping has 92 elements (counting the key if implicit)) 根本原因: 对于4000个字符,hive Metastore中SERDE_PARAMS表中PARAM_VALUE字段的字符限制是此问题的根本原因。 此限制可防止Hive创建具有高列数…

2020-09-23

insert into table ads_user_action_convert_day select ‘2019-02-10’, uv.day_count, ua.order_count, cast(ua.order_count/uv.day_count as decimal(10,2)) visitor2order_convert_ratio, ua.payment_count, cast(ua.payment_count/ua.order_count as decimal(10,2)) orde…

2020-09-28

Mybatis中#{}和${} 1、#{}将传入的数据都当成一个字符串,会对自动传入的数据加一个引号(单引号?双引号?加了引号就对了)如: //传入的值是sex order by #{column} 解析后为 order by “sex” //将会出错 2…

PageUtil

Data NoArgsConstructor public class PageUtil implements Serializable {private static final long serialVersionUID 1L;/*** 总记录数*/private int totalCount;/*** 每页记录数*/private int pageSize;/*** 总页数*/private int totalPage;/*** 当前页数*/private int c…

springboot+事务,多张表的操作事务回滚

第一步,在springboot的启动类上开启事务,注解 EnableTransactionManagement 第二步:事务注解,回滚 Transactional(rollbackFor Exception.class) //Transactional和try catch捕获异常会让注解失效,所以: 第…

@EnableTransactionManagement

GetMapping("/saveData")ApiOperation("传感器数据上云-红外测温及气体")Transactional(rollbackFor Exception.class)public WebResult saveData(InfrareDataReqVo infrareDataReqVo){try {PigHeatEntity pigHeatEntitynew PigHeatEntity();pigHeatEntit…

根据经纬度显示地图轨迹

后端代码 public boolean saveCarGPSData(CarGPSReqVo carGPSReqVo) {Map<String,String> paramsnew HashMap<>(2);params.put("key",trsConfig.getCarGPS_key());params.put("coorType",carGPSReqVo.getCoorType());params.put("pageNu…

Linux vim 全选

全选&#xff08;高亮显示&#xff09;&#xff1a;按esc后&#xff0c;然后ggvG或者ggVG 全部复制&#xff1a;按esc后&#xff0c;然后ggyG 全部删除&#xff1a;按esc后&#xff0c;然后dG