zoukankan      html  css  js  c++  java
  • Spark 的 Broadcast 和 Accumulator 的使用

    适合使用 Broadcast 的场景

    • 如果希望 driver 和每个 executor 读到的变量值是一致的
    • 如果有的变量存储的数据比较大,希望在每个 executor 都只初始化一遍

    broadcast 的变量是在 driver 初始化的,然后将 broadcast 变量的元数据传给 executor,当 executor 使用 broadcast 的时候先在本地查找,如果没有就从 driver 拉取一份,然后存在本地

    由于 broadcast 是在 driver 初始化并分发到各个 executor 的,并且 broadcast 是只读的,所以能保证一致性,即 driver 和各个 executor 读到的数据是一致的

    如果不用 broadcast 那么每次运行 task 都需要或者初始化变量,或者由 driver 启动 task 的时候传递变量,这样做一般没问题,但如果变量存储的数据比较大,效率就低了,使用了 broadcast 后可以保证每个 executor 只从 driver 拉取一次数据,后续的 task 可以重复使用

    例子

    val data = List(1, 2, 3, 4, 5, 6)
    val bdata = sc.broadcast(data)
    
    val rdd = sc.parallelize(1 to 6, 2)
    val observedSizes = rdd.map(_ => bdata.value.size)
    

    注意由于 driver 传到 executor 需要序列化,一些序列化不了的变量无法使用 broadcast


    如果需要跨多个 task 统计数据,就可以使用累加器 Accumulator

    例子

    // scala
    val accum = sc.longAccumulator("My Accumulator")   // Accumulator 可以有名字,在 Spark UI 可以看到
    sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
    accum.value
    
    # python
    accum = sc.accumulator(0)   # Accumulator 初始值是 0
    sc.parallelize([1, 2, 3, 4]).foreach(lambda x: accum.add(x))
    accum.value
    

    scala 和 python 的用法不一样



  • 相关阅读:
    彻底解决python cgi 编程出现的编码问题
    设置 mysql事物隔离级别
    python multiprocessing.Pool 中map、map_async、apply、apply_async的区别
    python 多线程、多进程、协程性能对比(以爬虫为例)
    一个学习git版本管理的超棒网站
    python3将unicode转化成中文输出
    python jieba包用法总结
    Oracle Dataguard
    Kubernetes -- DaemonSet
    STRIDE威胁建模
  • 原文地址:https://www.cnblogs.com/moonlight-lin/p/14166189.html
Copyright © 2011-2022 走看看