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 的用法不一样



  • 相关阅读:
    Hadoop学习笔记—20.网站日志分析项目案例(二)数据清洗
    python四舍五入保留2位小数
    查看python中的keywords(关键字)和modules
    c:forEach实现动态select标签
    解决Windows上数据库密码忘记问题
    Java读取properties文件总结
    URL地址最大长度问题
    Servlet生命周期
    解决mysql 数据库连接密码
    Java中int与integer的区别
  • 原文地址:https://www.cnblogs.com/moonlight-lin/p/14166189.html
Copyright © 2011-2022 走看看