// logRDD contains (userId -> ip), type is [Int, String]
// ips is memory hashmap, contains (ip -> provoince)

val ips=...  //内存Map，约100M
val rddResult=logRDD.Map{ p=>
val ip=p._2
val province=ips.getOrElse(ip,null)
(p._1,province)
}

val ips = … // ips是一个内存map，约100M
val bIPs= sc.broadcast(ips)
val resultRDD = logRDD.map { p =>
val ip = p._2
val provoince = bIPs.value.getOrElse(ip, null)
(p._1, provoince)
}