val a = 3 
val broadcast = sc.broadcast(a)  //把變量a廣播出去
val c = broadcast.value  //從廣播變量裏獲取a

// 推薦：使用不可變對象
case class Config(threshold: Int)  // val 不可變字段

val config = Config(100)
val broadcastConfig = sc.broadcast(config)

// 只能讀取，不能修改
rdd.map { data =>
  val currentThreshold = broadcastConfig.value.threshold  // ✅ 安全
  // ... 只讀操作
}

var currentConfig = Config(100)
var broadcastConfig = sc.broadcast(currentConfig)

// 需要更新時
currentConfig = Config(200)
broadcastConfig.unpersist()  // 取消舊的廣播
broadcastConfig = sc.broadcast(currentConfig)  // 廣播新對象

import org.apache.spark.sql.SparkSession

object BigRDDJoinSmallRDD {

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

    val sparkSession = SparkSession.builder().master("local[3]").appName("BigRDD Join SmallRDD").getOrCreate()
    val sc = sparkSession.sparkContext
    val list1 = List(("jame",23), ("wade",3), ("kobe",24))
    val list2 = List(("jame", 13), ("wade",6), ("kobe",16))
    val bigRDD = sc.makeRDD(list1)
    val smallRDD = sc.makeRDD(list2)

    println(bigRDD.getNumPartitions)
    println(smallRDD.getNumPartitions)

    // driver端rdd不broadcast廣播smallRDD到各executor，RDD不能被broadcast，需要轉換成數組array
    val  smallRDDB= sc.broadcast(smallRDD.collect())

    val joinedRDD = bigRDD.mapPartitions(partition => {
      val smallRDDBV = smallRDDB.value  // 各個executor端的task讀取廣播value
      partition.map(element => {
        //println(joinUtil(element, smallRDDBV))
        joinUtil(element, smallRDDBV)
      })
    })
    joinedRDD.foreach(x => println(x))



  }


/**
  * join操作：對兩個rdd中的相同key的value1和value2進行聚合，即(key,value1).join(key,value2)得到(key,(value1, vlaue2))
  * 如果bigRDDEle的key和smallRDD的某個key一致，那麼返回(key,(value1, vlaue2))
  * 該方法會在各executor的task上執行
  * */
 def joinUtil(bigRDDEle:(String,Int), smallRDD: Array[(String, Int)]): (String, (Int,Int)) = {
   var joinEle:(String, (Int, Int)) = null
   // 遍歷數組smallRDD
   smallRDD.foreach(smallRDDEle => {
      if(smallRDDEle._1.equals(bigRDDEle._1)){
        // 如果bigRDD中某個元素的key和數組smallRDD的key一致，返回join結果
        joinEle = (bigRDDEle._1, (bigRDDEle._2, smallRDDEle._2))
      }
    })
   joinEle
 }

}

val df: DataFrame = _
df.cache.count
 
val plan = df.queryExecution.logical
val estimated: BigInt = spark
.sessionState
.executePlan(plan)
.optimizedPlan
.stats
.sizeInBytes

val table1: DataFrame = spark.read.parquet(path1)
val table2: DataFrame = spark.read.parquet(path2)
table1.createOrReplaceTempView("t1")
table2.createOrReplaceTempView("t2")
 
val query: String = “select /*+ broadcast(t2) */ * from t1 inner join t2 on t1.key = t2.key”
val queryResutls: DataFrame = spark.sql(query)

table1.join(table2.hint(“broadcast”), Seq(“key”), “inner”)

import org.apache.spark.sql.functions.broadcast
table1.join(broadcast(table2), Seq(“key”), “inner”)

// 監控廣播 join 是否合適
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100MB")

val df = broadcast(mediumTable).join(largeTable, "id")

// 查看執行計劃
df.explain()

// 監控廣播數據大小
val broadcastSize = mediumTable.queryExecution.optimizedPlan.stats.sizeInBytes
println(s"Broadcast table size: ${broadcastSize / 1024 / 1024} MB")

// 如果看到以下情況，應考慮使用 shuffle：
// 1. GC 時間佔比高
// 2. 執行時間比 shuffle 還長
// 3. 頻繁的 spill 到磁盤

package com.sjf.open.spark;

import com.google.common.collect.Lists;
import org.apache.spark.util.AccumulatorV2;

import java.util.ArrayList;
import java.util.List;

/**
 * 自定義累加器 CollectionAccumulator
 * @author sjf0115
 * @Date Created in 下午2:11 18-6-4
 */
public class CollectionAccumulator<T> extends AccumulatorV2<T, List<T>> {

    private List<T> list = Lists.newArrayList();

    @Override
    public boolean isZero() {
        return list.isEmpty();
    }

    @Override
    public AccumulatorV2<T, List<T>> copy() {
        CollectionAccumulator<T> accumulator = new CollectionAccumulator<>();
        synchronized (accumulator) {
            accumulator.list.addAll(list);
        }
        return accumulator;
    }

    @Override
    public void reset() {
        list.clear();
    }

    @Override
    public void add(T v) {
        list.add(v);
    }

    @Override
    public void merge(AccumulatorV2<T, List<T>> other) {
        if(other instanceof CollectionAccumulator){
            list.addAll(((CollectionAccumulator) other).list);
        }
        else {
            throw new UnsupportedOperationException("Cannot merge " + this.getClass().getName() + " with " + other.getClass().getName());
        }
    }

    @Override
    public List<T> value() {
        return new ArrayList<>(list);
    }
}

package com.sjf.open.spark;

import com.google.common.collect.Lists;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;

import java.io.Serializable;
import java.util.List;

/**
 * 自定義累加器示例
 * @author sjf0115
 * @Date Created in 下午2:11 18-6-4
 */
public class CustomAccumulatorExample implements Serializable{

    public static void main(String[] args) {
        String appName = "CustomAccumulatorExample";
        SparkConf conf = new SparkConf().setAppName(appName);
        JavaSparkContext sparkContext = new JavaSparkContext(conf);

        List<String> list = Lists.newArrayList();
        list.add("27.34832,111.32135");
        list.add("34.88478,185.17841");
        list.add("39.92378,119.50802");
        list.add("94,119.50802");

        CollectionAccumulator<String> collectionAccumulator = new CollectionAccumulator<>();
        sparkContext.sc().register(collectionAccumulator, "Illegal Coordinates");
        // 原始座標
        JavaRDD<String> sourceRDD = sparkContext.parallelize(list);
        // 過濾非法座標
        JavaRDD<String> resultRDD = sourceRDD.filter(new Function<String, Boolean>() {
            @Override
            public Boolean call(String str) throws Exception {
                String[] coordinate = str.split(",");
                double lat = Double.parseDouble(coordinate[0]);
                double lon = Double.parseDouble(coordinate[1]);
                if(Math.abs(lat) > 90 || Math.abs(lon) > 180){
                    collectionAccumulator.add(str);
                    return true;
                }
                return false;
            }
        });
        // 輸出
        resultRDD.foreach(new VoidFunction<String>() {
            @Override
            public void call(String coordinate) throws Exception {
                System.out.println("[Data]" + coordinate);
            }
        });
        // 查看異常座標
        for (String coordinate : collectionAccumulator.value()) {
            System.out.println("[Illegal]: " + coordinate);
        }
    }

}

val acc = sc.longAccumulator("count")

// 創建 RDD
val rdd = sc.parallelize(1 to 10)

// transformation - 不會立即執行
val mappedRDD = rdd.map { x =>
  acc.add(1)
  x * 2
}

// 每次 action 都會導致 transformation 重新執行！
mappedRDD.count()  // acc = 10
mappedRDD.collect()  // acc = 20（又執行了一次！）

println(acc.value)  // 輸出 20，不是 10！