import spark.SparkContext
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import SparkContext._
import scala.collection.mutable.HashMapobject CSDNPassword {val log = LoggerFactory.getLogger("CSDNPassword");def main(args: Array[String]) {System.setProperty("spark.broadcast.factory", "spark.broadcast.BitTorrentBroadcastFactory")val spark = new SparkContext(args(0), "CSDNPassword", null, Array(args(1)))val slices = if (args.length > 2) args(2).toInt else 2log.info("spark.slices :" + slices)val file = spark.textFile(args(3), slices).filter(_.contains("@"))val passwords = file.map { user =>val user_s = user.split("#")val pw = user_s(1).trim()if (pw.length() > 5) (pw, 1) else ("Nil", 0)}val passwordLimit = if(args.length > 4) args(4).toInt else 1000val mailLimit = if(args.length > 5) args(5).toInt else 1000val passwordModel = passwords.reduceByKey(_ + _)val filePassoword = passwordModel.filter(_._2 > passwordLimit)//收集出现超过passwordLimit的密码val passwordCollect = filePassoword.collect();    /*** 白痴密码指数化，(count*1000)/max*/val zhishu = 1000val passwordMap = new HashMap[String, Int]()val sortPassword = passwordCollect.sortWith((A, B) => A._2 > B._2)val max = sortPassword(0)._2passwordMap += (sortPassword(0)._1 -> zhishu)for (i &lt;- 1 to sortPassword.length - 1) {       val weight = (sortPassword(i)._2 * zhishu) / max             passwordMap += (sortPassword(i)._1 -> weight)}//包装成广播变量, 提供给所有work使用val passwordMapBro = spark.broadcast(passwordMap)log.info("---------------------create broadcast-------------------------")/val users2 = file.map{ user =>val user_s = user.split("#")val pw = user_s(1).trim()val weight = passwordMapBro.value.getOrElse(pw, 0)//得到密码白痴程序val user2_s = user.split("@")if (pw.length() > 5 && user2_s.length > 1) (user2_s(1).toLowerCase(), (1, weight)) else ("Nil",(0,  0))}val user2ReduceByKey = users2.reduceByKey{(a,b)=>(a._1 + b._1, a._2 + b._2)}val yes = user2ReduceByKey.filter(_._2._1 > mailLimit).map{user =>  //过滤出超过mailLimit的邮箱(user._1, user._2._2/user._2._1) //除以基数}//收集，排序val arrays = yes.collect()val sortMail = arrays.sortWith((A, B) => A._2 > B._2)val len = if(sortMail.length > 100) 100 else sortMail.lengthfor (i <- 0 to len -1) {log.info(i + " mail:" + sortMail(i)._1 + ", 密码白痴程度:" + sortMail(i)._2)}spark.stop()}
}