大数据开发之Spark(累加器、广播变量、Top10热门品类实战)

这篇具有很好参考价值的文章主要介绍了大数据开发之Spark(累加器、广播变量、Top10热门品类实战)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

第 3 章:累加器

累加器:分布式共享只写变量。(executor和executor之间不能读数据)
累加器用来把executor端变量信息聚合到driver端。在driver中定义的一个变量,在executor端的每个task都会得到这个变量的一份新的副本,每个task更新这些副本的值后,传回driver端进行合并计算。
大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

1、累加器使用
1)累加器定义(sparkcontext.accumulator(initialvalue)方法)

val sum: LongAccumulator = sc.longAccumulator("sum")

2)累加器添加数据(累加器.add方法)

sum.add(count)

3)累加器获取数据(累加器.value)

sum.value

2、创建包名:com.atguigu.accumulator
3、代码实现

object accumulator01_system {
package com.atguigu.cache
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object accumulator01_system {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]") 
val sc = new SparkContext(conf)

    val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
    //需求:统计a出现的所有次数 ("a",10)

    //普通算子实现 reduceByKey 代码会走shuffle 效率低
    //val rdd1: RDD[(String, Int)] = dataRDD.reduceByKey(_ + _)

//普通变量无法实现 
//结论:普通变量只能从driver端发给executor端,在executor计算完以后,结果不会返回给driver端
/*
    var sum = 0

    dataRDD.foreach{
      case (a,count) => {
        sum += count
        println("sum = " + sum)
      }
    }

    println(("a",sum))
*/
    //累加器实现
    //1 声明累加器
    val accSum: LongAccumulator = sc.longAccumulator("sum")

    dataRDD.foreach{
      case (a,count) => {
        //2 使用累加器累加  累加器.add()
        accSum.add(count)
        // 4 不要在executor端获取累加器的值,因为不准确 
//因此我们说累加器叫分布式共享只写变量
        //println("sum = " + accSum.value)
      }
    }
    //3 获取累加器的值 累加器.value
    println(("a",accSum.value))

    sc.stop()
  }
}

注意:executor端的任务不能读取累加器的值(例如:在executor端调用sum.value,获取的值不是累加器最终的值)。因此我们说,累加器事一个分布式共享只写变量。累加器要放在行动算子中。因为转换算子执行的次数取决于job的数量,如果一个spark应用有多个行动算子,那么转换算子中的累加器可能会不止一次更新,导致结果错误。所以,如果想要一个无论是失败还是重复计算时都绝对可靠的累加器,我们必须把它放在foreach()这样的行动算子中。
对于在行动算子中使用的累加器,spark只会把每个job对各累加器的修改应用一次。

object accumulator02_updateCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
    val sc = new SparkContext(conf)

    val dataRDD: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("a", 3), ("a", 4)))
    //需求:统计a出现的所有次数 ("a",10)
    //累加器实现
    //1 声明累加器
    val accSum: LongAccumulator = sc.longAccumulator("sum")

    val mapRDD: RDD[Unit] = dataRDD.map {
      case (a, count) => {
        //2 使用累加器累加  累加器.add()
        accSum.add(count)
        // 4 不要在executor端获取累加器的值,因为不准确 因此我们说累加器叫分布式共享只写变量
        //println("sum = " + accSum.value)
      }
    }

    //调用两次行动算子,map执行两次,导致最终累加器的值翻倍
    mapRDD.collect()
    mapRDD.collect()

    /**
     * 结论:使用累加器最好要在行动算子中使用,因为行动算子只会执行一次,而转换算子的执行次数不确定!
     */ 
    //2 获取累加器的值 累加器.value
    println(("a",accSum.value))
    
    sc.stop()
  }
}

一般在开发中使用的累加器为集合累加器,在某些场景可以减少shuffle
现在我们用集合累加器实现wordcount:

object Test02_Acc {
  def main(args: Array[String]): Unit = {
    //1、创建sparkcontext配置
	val conf = new SparkConf().setMaster("local[4]").setAppName("test")
	//2、创建sparkcontext
	val sc = new SparkContext(conf)
	//3、创建集合累加器,累加元素为Map
	val acc = sc.collectionAccumulator[mutable.Map[String,Int]]
	//4、读取文件
	val rdd1 = sc.textFile("datas/wc.txt")
	//5、切割+转换
	val rdd2 = rdd1.flatMap(x=>x.split(" "))
	//6、转换为KV键值对
	val rdd3 = rdd2.map(x=>(x,1))
	//7、使用foreachPartitions在每个分区中对所有单词累加,将累加结果放入累加器中
	rdd3.foreachPartition(it=> {
  		//创建一个累加Map容器
  		val map = mutable.Map[String,Int]()
  		//遍历分区数据
  		it.foreach(x=>{
    		val num = map.getOrElse(x._1,0)
    		//将单词累加到map容器中
    		map.put(x._1,num+x._2)
  		})
  		//将装载分区累加结果的map容器放入累加器中
  		acc.add(map)

	})
	//为了方便操作,将java集合转成scala集合
	import scala.collection.JavaConverters._
	//获取累加器结果,此时List中的每个Map是之前放入累加器的分区累加结果Map
	val r = acc.value.asScala
	//压平,将所有分区计算结果放入List中
	val pList = r.flatten
	//按照单词分组
	val rMap = pList.groupBy(x=>x._1)
	//,统计每个单词总个数
	val result = rMap.map(x => (x._1, x._2.map(_._2).sum))

	println(result)

    // 4.关闭sc
    sc.stop()
  }
}

第 4 章:广播变量
广播变量:分布式共享只读变量
广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值,以供一个或多个spark task操作使用。比如,如果你的应用需要向所有节点发送一个较大的只读查询表,广播变量会用起来会很顺手。在多个task并行操作中使用同一个变量,但是spark会为每个task任务分别发送。
1、使用广播变量步骤:
1)调用sparkcontext.broadcast(广播变量)创建出一个广播对象,任何可序列化的类型都可以这么实现。
2)通过广播变量.value,访问该对象的值。
3)广播变量只会被发到各个节点一次,作为只读值处理(修改这个值不会影响到别的节点)。
2、原理说明
大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

3、创建包名:com.atguigu.broadcast
4、代码实现

object broadcast01 {

    def main(args: Array[String]): Unit = {

        //1.创建SparkConf并设置App名称
        val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

        //2.创建SparkContext,该对象是提交Spark App的入口
        val sc: SparkContext = new SparkContext(conf)

        //3.创建一个字符串RDD,过滤出包含WARN的数据
        val rdd: RDD[String] = sc.makeRDD(List("WARN:Class Not Find", "INFO:Class Not Find", "DEBUG:Class Not Find"), 4)
        val str: String = "WARN"

        // 声明广播变量
        val bdStr: Broadcast[String] = sc.broadcast(str)

        val filterRDD: RDD[String] = rdd.filter {
            // log=>log.contains(str)
            log => log.contains(bdStr.value)
        }

        filterRDD.foreach(println)

        //4.关闭连接
        sc.stop()
    }
}

第 5 章:sparkcore实战

5.1 数据准备

1、数据格式
大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

1)数据采用_分割字段
2)每一行表示用户的一个行为,所以每一行只能是四种行为中的一种
3)如果点击的品类id和产品id是-1表示这次不是点击
4)针对下单行为,一次可以下单多个产品,所以品类id和产品id都是多个,id之间使用逗号分割。如果本次不是下单行为,则他们相关数据用null来表示。
5)支付行为和下单行为格式类似。
2、数据详情字段说明
大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

5.2 需求:top10热门品类

大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

需求说明:品类是指产品的分类,大型电商网站品类分多级,咱们的项目中品类只有一级,不同的公司可能对热门的定义不一样。我们按照每个品类的点击、下单、支付的量(次数)来统计热门品类。
大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

例如:综合排名=点击数20%+下单数30%+支付书数*50%
本项目需求优化为:先按照点击数排名,考前的就排名高;如果点击数相同,再比较下单数;下单数相同,就比较支付数。

5.2.1 需求分析(方案一)常规算子

思路:分别统计每个品类点击的次数,下单的次数和支付的次数。然后想办法将三个rdd联合在一块。
(品类,点击总数)(品类,下单总数)(品类,支付总数)
(品类,(点击总数,下单总数,支付总数))
然后就可以按照各品类的元组(点击总数,下单总数,支付总数)进行倒叙排序了,因为元组排序刚好是先排第一个元素,然后排第二个元素,最后第三个元素。最后取top10即可。

5.2.2 需求实现(方案一)

1)创建包名:com.atguigu.project01
2)方案一:代码实现(cogroup算子实现满外连接)

package com.atguigu.spark.demo

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Test01_Top10 {
  def main(args: Array[String]): Unit = {
    // 1. 创建配置对象
    val conf: SparkConf = new SparkConf().setAppName("sparkCore").setMaster("local[*]")

    // 2. 初始化sc
    val sc = new SparkContext(conf)

    // 3. 编写代码
    // 需求: top10的热门品类  通过(id,(点击,下单,支付))

    val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")

    // 1. 过滤点击数据   进行统计
    val clickRDD: RDD[String] = lineRDD.filter(line => {
      val data: Array[String] = line.split("_")
      // 读进来的数据全部都是字符串
      data(6) != "-1"
    })

    // 统计品类点击次数即可
    val clickCountRDD: RDD[(String, Int)] = clickRDD.map(line => {
      val data: Array[String] = line.split("_")
      (data(6), 1)
    })
      .reduceByKey(_ + _)

    // 2.  过滤统计下单品类
    val orderRDD: RDD[String] = lineRDD.filter(line => {
      val data: Array[String] = line.split("_")
      data(8) != "null"
    })

    val orderCountRDD: RDD[(String, Int)] = orderRDD.flatMap(line => {
      // 切分整行数据
      val data: Array[String] = line.split("_")
      // 切分下单品类
      val orders: Array[String] = data(8).split(",")
      // 改变数据结构  (下单品类,1)
      orders.map((_, 1))
    }).reduceByKey(_ + _)

    // 3. 过滤统计支付品类数据
    val payRDD: RDD[String] = lineRDD.filter(line => {
      val data: Array[String] = line.split("_")
      data(10) != "null"
    })

    val payCountRDD: RDD[(String, Int)] = payRDD.flatMap(line => {
      val data: Array[String] = line.split("_")
      val pays: Array[String] = data(10).split(",")
      pays.map((_, 1))
    }).reduceByKey(_ + _)


    // 使用cogroup满外连接  避免使用join出现有的品类只有点击没有下单支付  造成数据丢失
    val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] = clickCountRDD.cogroup(orderCountRDD, payCountRDD)

    // 改变数据结构  (id,(list(所有当前id的点击数据),list(所有当前id的下单数据),list(所有当前id的支付数据)))
    val cogroupRDD2: RDD[(String, (Int, Int, Int))] = cogroupRDD.mapValues({
      case (clickList, orderList, payList) => (clickList.sum, orderList.sum, payList.sum)
    })

    // 排序取top10
    val result: Array[(String, (Int, Int, Int))] = cogroupRDD2.sortBy(_._2, false).take(10)

    result.foreach(println)


    Thread.sleep(600000)

    // 4.关闭sc
    sc.stop()
  }
}

3)一次计算,转换数据结构,通过位置标记数据的类型,不再使用三次过滤,减少reducebykey的次数

package com.atguigu.spark.demo

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Test02_Top10 {
  def main(args: Array[String]): Unit = {
    // 1. 创建配置对象
    val conf: SparkConf = new SparkConf().setAppName("sparkCore").setMaster("local[*]")

    // 2. 初始化sc
    val sc = new SparkContext(conf)

    // 3. 编写代码
    // 需求: top10的热门品类  通过(id,(点击,下单,支付))

    val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")

    // 通过位置来标记属于什么类型的数据
//    val filterRDD: RDD[String] = lineRDD.filter(line => {
//      val data: Array[String] = line.split("_")
//      // 过滤出所有的点击  下单  支付数据
//      data(6) != "-1" || data(8) != "null" || data(10) != "null"
//    })

    // 转换数据结构
    // 1. 点击数据  -> (id,(1,0,0))
    // 2. 下单数据  -> (id,(0,1,0))
    // 3. 支付数据  -> (id,(0,0,1))
    val flatMapRDD: RDD[(String, (Int, Int, Int))] = lineRDD.flatMap(line => {
      val data: Array[String] = line.split("_")
      // 判断属于三种的哪一种
      if (data(6) != "-1") {
        // 点击数据
        List((data(6), (1, 0, 0)))
      } else if (data(8) != "null") {
        // 下单数据
        // 此处为数组  需要拆分为多个
        val orders: Array[String] = data(8).split(",")
        // 多个订单(15,13,5) =>  (15,(0,1,0)),(13,(0,1,0))...
        orders.map(order => (order, (0, 1, 0)))

      } else if (data(10) != "null") {
        // 支付数据
        val pays: Array[String] = data(10).split(",")
        pays.map(pay => (pay, (0, 0, 1)))
      } else {
        List()
      }
    })

    val reduceRDD: RDD[(String, (Int, Int, Int))] = flatMapRDD.reduceByKey((res, elem) => (res._1 + elem._1, res._2 + elem._2, res._3 + elem._3))

    val result: Array[(String, (Int, Int, Int))] = reduceRDD.sortBy(_._2, false).take(10)

    result.foreach(println)

    Thread.sleep(600000)

    // 4.关闭sc
    sc.stop()
  }
}

5.2.3 需求分析(方案二)样例类

使用样例类的方式实现。
大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

5.2.4 需求实现(方案二)

1、用来封装用户行为的样例类

//用户访问动作表
case class UserVisitAction(date: String,//用户点击行为的日期
                           user_id: String,//用户的ID
                           session_id: String,//Session的ID
                           page_id: String,//某个页面的ID
                           action_time: String,//动作的时间点
                           search_keyword: String,//用户搜索的关键词
                           click_category_id: String,//某一个商品品类的ID
                           click_product_id: String,//某一个商品的ID
                           order_category_ids: String,//一次订单中所有品类的ID集合
                           order_product_ids: String,//一次订单中所有商品的ID集合
                           pay_category_ids: String,//一次支付中所有品类的ID集合
                           pay_product_ids: String,//一次支付中所有商品的ID集合
                           city_id: String)//城市 id
// 输出结果表
case class CategoryCountInfo(categoryId: String,//品类id
                             clickCount: Long,//点击次数
                             orderCount: Long,//订单次数
                             payCount: Long)//支付次数
注意:样例类的属性默认是val修饰,不能修改;需要修改属性,需要采用var修饰。
// 输出结果表
case class CategoryCountInfo(var categoryId: String,//品类id
                             var clickCount: Long,//点击次数
                             var orderCount: Long,//订单次数
                             var payCount: Long)//支付次数

注意:样例类的属性默认是val修饰,不能修改;需要修改属性,需要采用var修饰。

2、核心业务代码实现

package com.atguigu.spark.demo

import com.atguigu.spark.demo.bean.{CategoryCountInfo, UserVisitAction}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Test04_Top10 {
  def main(args: Array[String]): Unit = {
    // 1. 创建配置对象
    val conf: SparkConf = new SparkConf().setAppName("sparkCore").setMaster("local[*]")

    // 2. 初始化sc
    val sc = new SparkContext(conf)

    // 3. 编写代码
    // 读取数据
    val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")

    // 转换为样例类
    val userRDD: RDD[UserVisitAction] = lineRDD.map(line => {
      val data: Array[String] = line.split("_")
      UserVisitAction(
        data(0),
        data(1),
        data(2),
        data(3),
        data(4),
        data(5),
        data(6),
        data(7),
        data(8),
        data(9),
        data(10),
        data(11),
        data(12)
      )
    })


    //切分数据为单个品类
    val categoryRDD: RDD[CategoryCountInfo] = userRDD.flatMap(user => {
      if (user.click_category_id != "-1") {
        // 点击数据
        List(CategoryCountInfo(user.click_category_id, 1, 0, 0))
      } else if (user.order_category_ids != "null") {
        // 下单数据
        val orders: Array[String] = user.order_category_ids.split(",")
        orders.map(order => CategoryCountInfo(order, 0, 1, 0))

      } else if (user.pay_category_ids != "null") {
        // 支付数据
        val pays: Array[String] = user.pay_category_ids.split(",")
        pays.map(pay => CategoryCountInfo(pay, 0, 0, 1))

      } else {
        List()
      }
    })

    // 聚合同一品类的数据
    val groupRDD: RDD[(String, Iterable[CategoryCountInfo])] = categoryRDD.groupBy(_.categoryId)


    val value: RDD[(String, CategoryCountInfo)] = groupRDD.mapValues(list => {
      // 集合常用函数
      list.reduce((res, elem) => {
        res.clickCount += elem.clickCount
        res.orderCount += elem.orderCount
        res.payCount += elem.payCount
        res
      })
    })

    val categoryReduceRDD: RDD[CategoryCountInfo] = value.map(_._2)

    // 排序取top10
    val result: Array[CategoryCountInfo] = categoryReduceRDD.sortBy(info =>
      (info.clickCount, info.orderCount, info.payCount),false)
      .take(10)

    result.foreach(println)

    Thread.sleep(600000)

    // 4.关闭sc
    sc.stop()
  }
}

5.2.5 需求分析(方案三)样例类+算子优化

针对方案二中的groupby算子,没有提前聚合的功能,替换成reducebykey
大数据开发之Spark(累加器、广播变量、Top10热门品类实战),大数据,spark,分布式

5.2.6 需求实现(方案三)

1、样例类代码和方案二一样。
2、核心代码实现文章来源地址https://www.toymoban.com/news/detail-822041.html

package com.atguigu.spark.demo

import com.atguigu.spark.demo.bean.{CategoryCountInfo, UserVisitAction}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Test05_Top10 {
  def main(args: Array[String]): Unit = {
    // 1. 创建配置对象
    val conf: SparkConf = new SparkConf().setAppName("sparkCore").setMaster("local[*]")

    // 2. 初始化sc
    val sc = new SparkContext(conf)

    // 3. 编写代码
    // 读取数据
    val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")

    // 转换为样例类
    val userRDD: RDD[UserVisitAction] = lineRDD.map(line => {
      val data: Array[String] = line.split("_")
      UserVisitAction(
        data(0),
        data(1),
        data(2),
        data(3),
        data(4),
        data(5),
        data(6),
        data(7),
        data(8),
        data(9),
        data(10),
        data(11),
        data(12)
      )
    })


    //切分数据为单个品类
    val categoryRDD: RDD[CategoryCountInfo] = userRDD.flatMap(user => {
      if (user.click_category_id != "-1") {
        // 点击数据
        List(CategoryCountInfo(user.click_category_id, 1, 0, 0))
      } else if (user.order_category_ids != "null") {
        // 下单数据
        val orders: Array[String] = user.order_category_ids.split(",")
        orders.map(order => CategoryCountInfo(order, 0, 1, 0))

      } else if (user.pay_category_ids != "null") {
        // 支付数据
        val pays: Array[String] = user.pay_category_ids.split(",")
        pays.map(pay => CategoryCountInfo(pay, 0, 0, 1))

      } else {
        List()
      }
    })

    // 聚合同一品类的数据
    // 使用reduceByKey调换groupBy  (重要)
    val reduceRDD: RDD[(String, CategoryCountInfo)] = categoryRDD.map(info => (info.categoryId, info))
      .reduceByKey((res, elem) => {
        res.clickCount += elem.clickCount
        res.orderCount += elem.orderCount
        res.payCount += elem.payCount
        res
      })

    val categoryReduceRDD: RDD[CategoryCountInfo] = reduceRDD.map(_._2)

    // 排序取top10
    val result: Array[CategoryCountInfo] = categoryReduceRDD.sortBy(info =>
      (info.clickCount, info.orderCount, info.payCount),false)
      .take(10)

    result.foreach(println)


    Thread.sleep(600000)

    // 4.关闭sc
    sc.stop()
  }
}

到了这里,关于大数据开发之Spark(累加器、广播变量、Top10热门品类实战)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spark---累加器

    累加器用来把 Executor 端变量信息聚合到 Driver 端。在 Driver 程序中定义的变量,在Executor 端的每个 Task 都会得到这个变量的一份新的副本,每个 task 更新这些副本的值后,传回 Driver 端进行 merge。 运行结果: 我们预期是想要实现数据的累加,开始数据从Driver被传输到了Execut

    2024年02月02日
    浏览(46)
  • Spark累加器LongAccumulator

    1.Accumulator是由Driver端总体进行维护的,读取当前值也是在Driver端,各个Task在其所在的Executor上也维护了Accumulator变量,但只是局部性累加操作,运行完成后会到Driver端去合并累加结果。Accumulator有两个性质: 1、只会累加,合并即累加; 2、不改变Spark作业懒执行的特点,即没

    2024年01月25日
    浏览(53)
  • 【Spark原理系列】Accumulator累加器原理用法示例源码详解

    源自专栏《SparkML:Spark ML系列专栏目录》 Accumulator是Spark中的一种分布式变量,用于在并行计算中进行累加操作。它是由MapReduce模型中的“全局计数器”概念演化而来的。 Accumulator提供了一个可写的分布式变量,可以在并行计算中进行累加操作。在Spark中,当一个任务对Accum

    2024年03月14日
    浏览(62)
  • Flink 源码剖析|累加器

    累加器是实现了 加法运算 功能和 合并运算 (合并多个累加器的结果)功能的一种数据结构,在作业结束后,可以获取所有部分(各个 operator 的各个 subtask)合并后的最终结果并发送到客户端。 Flink 的累加器均实现了 Accumulator 接口,包括如下 2 个方法用于支持加法运算和合

    2024年02月21日
    浏览(53)
  • 计算机组成原理 累加器实验

    计算机组成原理实验环境 理解累加器的概念和作用。 连接运算器、存储器和累加器,熟悉计算机的数据通路。 掌握使用微命令执行各种操作的方法。 做好实验预习,读懂实验电路图,熟悉实验元器件的功能特性和使用方法。在实验之前设计好要使用的微命令,填入表 6-2 、

    2024年02月06日
    浏览(41)
  • Flink 源码剖析|4. 累加器与相关工具方法

    累加器是实现了 加法运算 功能和 合并运算 (合并多个累加器的结果)功能的一种数据结构,在作业结束后,可以获取所有部分(各个 operator 的各个 subtask)合并后的最终结果并发送到客户端。 Flink 的累加器均实现了 Accumulator 接口,包括如下 2 个方法用于支持加法运算和合

    2024年03月15日
    浏览(45)
  • 【数字IC/FPGA】百度昆仑芯手撕代码--累加器

    已知一个加法器IP,其功能是计算两个数的和,但这个和延迟两个周期才会输出。现在有一串连续的数据输入,每个周期都不间断,试问最少需要例化几个上述的加法器IP,才可以实现累加的功能。 由于加法器两个周期后才能得到结果(再将该结果作为加法器的输入进行累加

    2024年02月09日
    浏览(41)
  • 《JUC并发编程 - 高级篇》05 -共享模型之无锁 (CAS | 原子整数 | 原子引用 | 原子数组 | 字段更新器 | 原子累加器 | Unsafe类 )

    有如下需求,保证 account.withdraw 取款方法的线程安全 原有实现并不是线程安全的 测试代码 执行测试代码,某次执行结果 5.1.1 为么不安全 withdraw 方法是临界区,会存在线程安全问题 查看下字节码 多线程在执行过程中可能会出现指令的交错,从而结果错误! 5.1.2 解决思路1

    2023年04月12日
    浏览(46)
  • spark广播变量

    2024-1-24 广播变量特点 Broadcast Variable会将使用到的变量,只会为每个节点拷贝一份,不会为每个task进行拷贝,能够优化性能(在task数量比较大体现更明显),减少网络传输及内存消耗 通过SparkContext的broadcast()方法,针对某个变量创建广播变量,可以通过广播变量的value()方法

    2024年01月25日
    浏览(69)
  • Apache Spark中的广播变量分发机制

    Apache Spark中的广播变量提供了一种机制,允许用户在集群中共享只读变量,并且每个任务都可以访问这个变量,而不需要在每次任务之间重新发送该变量。这种机制特别适用于在所有节点上都需要访问同一份只读数据集的情况,因为它可以显著减少网络通信的开销。 以下是广

    2024年01月24日
    浏览(50)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包