【GitHub】Spark with Scala & Alogorithm

相似文件 换一批

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

1. Cake Pattern
   
2. 
   
3. Worker.scala
   
4. master = context.actorSelection(Master.toAkkaUrl(activeMasterUrl))
   
5. 
   
6. 
   
7. ActorRefProvider.scala
   
8. trait ActorRefFactory {
   
9. def actorSelection(path: String): ActorSelection = path match {
   
10.   case RelativeActorPath(elems) ⇒
    
11.     if (elems.isEmpty) ActorSelection(provider.deadLetters, "")
    
12.     else if (elems.head.isEmpty) ActorSelection(provider.rootGuardian, elems.tail)
    
13.     else ActorSelection(lookupRoot, elems)
    
14.   case ActorPathExtractor(address, elems) ⇒
    
15.     ActorSelection(provider.rootGuardianAt(address), elems)
    
16.   case _ ⇒
    
17.     ActorSelection(provider.deadLetters, "")
    
18. }

复制代码

本帖隐藏的内容

spark-with-scala.pdf (1.01 MB)

2017-2-21 10:02:03 上传

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：address

1. Extractor Objects
   
2. 
   
3. 参见 官方文档 Extractor Objects
   
4. 
   
5. 下面Spark SQL代码，当遇到  case ExtractEquiJoinKeys 时自动使用ExtractEquiJoinKeys对象(?only object) unapply 方法
   
6.       case ExtractEquiJoinKeys(Inner, leftKeys, rightKeys, condition, left, right) =>
   
7.         val buildSide =
   
8.           if (right.statistics.sizeInBytes <= left.statistics.sizeInBytes) {
   
9.             joins.BuildRight
   
10.           } else {
    
11.             joins.BuildLeft
    
12.           }
    
13.         val hashJoin = joins.ShuffledHashJoin(
    
14.           leftKeys, rightKeys, buildSide, planLater(left), planLater(right))
    
15.         condition.map(Filter(_, hashJoin)).getOrElse(hashJoin) :: Nil
    
16. ...  
    
17. 
    
18. object ExtractEquiJoinKeys extends Logging with PredicateHelper {
    
19.   /** (joinType, rightKeys, leftKeys, condition, leftChild, rightChild) */
    
20.   type ReturnType =
    
21.     (JoinType, Seq[Expression], Seq[Expression], Option[Expression], LogicalPlan, LogicalPlan)
    
22. 
    
23.   def unapply(plan: LogicalPlan): Option[ReturnType] = plan match {
    
24.     case join @ Join(left, right, joinType, condition) =>
    
25.       logDebug(s"Considering join on: $condition")
    
26.       // Find equi-join predicates that can be evaluated before the join, and thus can be used
    
27.       // as join keys.
    
28.       val (joinPredicates, otherPredicates) =
    
29.         condition.map(splitConjunctivePredicates).getOrElse(Nil).partition {
    
30.           case EqualTo(l, r) if (canEvaluate(l, left) && canEvaluate(r, right)) ||
    
31.             (canEvaluate(l, right) && canEvaluate(r, left)) => true
    
32.           case _ => false
    
33.         }
    
34. 
    
35.       val joinKeys = joinPredicates.map {
    
36.         case EqualTo(l, r) if canEvaluate(l, left) && canEvaluate(r, right) => (l, r)
    
37.         case EqualTo(l, r) if canEvaluate(l, right) && canEvaluate(r, left) => (r, l)
    
38.       }
    
39.       val leftKeys = joinKeys.map(_._1)
    
40.       val rightKeys = joinKeys.map(_._2)
    
41. 
    
42.       if (joinKeys.nonEmpty) {
    
43.         logDebug(s"leftKeys:$leftKeys | rightKeys:$rightKeys")
    
44.         Some((joinType, leftKeys, rightKeys, otherPredicates.reduceOption(And), left, right))
    
45.       } else {
    
46.         None
    
47.       }
    
48.     case _ => None
    
49.   }

复制代码

1. Try Some None
   
2. 
   
3. HadoopRDD.scala
   
4.   override def getPreferredLocations(split: Partition): Seq[String] = {
   
5.     val locs: Option[Seq[String]] = HadoopRDD.SPLIT_INFO_REFLECTIONS match {
   
6.       case Some(c) =>
   
7.         try {
   
8.           val lsplit = c.inputSplitWithLocationInfo.cast(hsplit)
   
9.           val infos = c.getLocationInfo.invoke(lsplit).asInstanceOf[Array[AnyRef]]
   
10.           Some(HadoopRDD.convertSplitLocationInfo(infos))
    
11.         } catch {
    
12.           case e: Exception =>
    
13.             logDebug("Failed to use InputSplitWithLocations.", e)
    
14.             None
    
15.         }
    
16.       case None => None
    
17.     }
    
18.     locs.getOrElse(hsplit.getLocations.filter(_ != "localhost"))
    
19.   }
    
20. 
    
21. 
    
22. ??： 如果不写HadoopRDD.SPLIT... 则找不到SPLIT_INFO_REFLECTIONS
    
23. 
    
24. locs有可能是Some或者None, 则getOrElse，传进去一个函数，即去掉本地节点
    
25. private[spark] val SPLIT_INFO_REFLECTIONS: Option[SplitInfoReflections] = try {
    
26.     Some(new SplitInfoReflections)
    
27.   } catch {
    
28.     case e: Exception =>
    
29.       logDebug("SplitLocationInfo and other new Hadoop classes are " +
    
30.           "unavailable. Using the older Hadoop location info code.", e)
    
31.       None
    
32.   }
    
33. 
    
34.   private[spark] class SplitInfoReflections {
    
35.     val inputSplitWithLocationInfo =
    
36.       Class.forName("org.apache.hadoop.mapred.InputSplitWithLocationInfo")
    
37.     val getLocationInfo = inputSplitWithLocationInfo.getMethod("getLocationInfo")
    
38.     val newInputSplit = Class.forName("org.apache.hadoop.mapreduce.InputSplit")
    
39.     val newGetLocationInfo = newInputSplit.getMethod("getLocationInfo")
    
40.     val splitLocationInfo = Class.forName("org.apache.hadoop.mapred.SplitLocationInfo")
    
41.     val isInMemory = splitLocationInfo.getMethod("isInMemory")
    
42.     val getLocation = splitLocationInfo.getMethod("getLocation")
    
43.   }

复制代码

1. Iterator
   
2. 
   
3. DAGScheduler.scala
   
4. def submitJob[T, U](
   
5.       rdd: RDD[T],
   
6.       func: (TaskContext, Iterator[T]) => U,
   
7.       partitions: Seq[Int],
   
8.       callSite: CallSite,
   
9.       allowLocal: Boolean,
   
10.       resultHandler: (Int, U) => Unit,
    
11.       properties: Properties = null): JobWaiter[U] =
    
12.   {
    
13. 
    
14.     partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
    
15.       throw new IllegalArgumentException(
    
16.         "Attempting to access a non-existent partition: " + p + ". " +
    
17.           "Total number of partitions: " + maxPartitions)
    
18.     }

复制代码

1. dropWhile
   
2. //RDDOperationScope.withScope
   
3. val callerMethodName = Thread.currentThread.getStackTrace()
   
4.       .dropWhile(_.getMethodName != ourMethodName)
   
5.       .find(_.getMethodName != ourMethodName)
   
6.       .map(_.getMethodName)
   
7.       .getOrElse {
   
8.         // Log a warning just in case, but this should almost certainly never happen
   
9.         logWarning("No valid method name for this RDD operation scope!")
   
10.         "N/A"
    
11.       }
    
12. 
    
13. 
    
14. dropWhile(p) 表示如果集合元素符合条件则略过 如
    
15. 
    
16. +
    
17. val it = Iterator("a", "number", "of", "words")
    
18. //定义了小于2的忽略
    
19. it dropWhile (_.length < 2)
    
20. it.next()
    
21. 
    
22. 
    
23. 当执行 it.next 到达"a"时忽略，直接到达"number"

复制代码

谢谢楼主分享！

谢谢分享

谢谢分享