tar zxvf scala-2.10.6.tgz -C /opt/app

vi /etc/profile

export SCALA_HOME=/opt/app/scala-2.10.6export PATH=$SCALA_HOME/bin:$PATH

source /etc/profile # 生效env | grep SCALA_HOME # 验证

tar zxvf spark-1.6.3-bin-hadoop2.6.tgz -C /opt/appcd /opt/app/spark-1.6.3-bin-hadoop2.6/confcp spark-env.sh.template spark-env.shvi spark-env.sh

export JAVA_HOME=/opt/app/jdk1.8.0_121export SCALA_HOME=/opt/app/scala-2.10.6export HADOOP_HOME=/opt/app/hadoop-2.6.5export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoopexport YARN_CONF_DIR=${HADOOP_HOME}/etc/hadoop# For standalone modeexport SPARK_WORKER_CORES=1export SPARK_DAEMON_MEMORY=512m

cp spark-defaults.conf.template spark-defaults.confhadoop fs -mkdir /spark.eventLog.dirvi spark-defaults.conf

spark.driver.extraClassPath        /opt/app/apache-hive-1.2.2-bin/lib/mysql-connector-java-5.1.22-bin.jarspark.eventLog.enabled           truespark.eventLog.dir               hdfs://centos1:9000/spark.eventLog.dir

cp slaves.template slavesvi slaves

centos2centos3

ln -s /opt/app/apache-hive-1.2.2-bin/conf/hive-site.xml .

scp -r /opt/app/spark-1.6.3-bin-hadoop2.6 hadoop@centos2:/opt/appscp -r /opt/app/spark-1.6.3-bin-hadoop2.6 hadoop@centos3:/opt/app

export SPARK_HOME=/opt/app/spark-1.6.3-bin-hadoop2.6export PATH=$PATH:$SPARK_HOME/bin

sbin/start-all.shjps

Master # Master机器进程Worker # Slave机器进程

bin/spark-submit --master spark://centos1:7077 --deploy-mode client --class org.apache.spark.examples.SparkPi --driver-memory 512m --executor-memory 512m --num-executors 1 --executor-cores 1 lib/spark-examples-1.6.3-hadoop2.6.0.jar # Standalone Client模式运行bin/spark-submit --master spark://centos1:7077 --deploy-mode cluster --class org.apache.spark.examples.SparkPi --driver-memory 512m --executor-memory 512m --num-executors 1 --executor-cores 1 lib/spark-examples-1.6.3-hadoop2.6.0.jar # Standalone Cluster模式运行bin/spark-submit --master yarn-client --class org.apache.spark.examples.SparkPi --driver-memory 512m --executor-memory 512m --num-executors 1 --executor-cores 1 lib/spark-examples-1.6.3-hadoop2.6.0.jar # Yarn Client模式运行bin/spark-submit --master yarn-cluster --class org.apache.spark.examples.SparkPi --driver-memory 512m --executor-memory 512m --num-executors 1 --executor-cores 1 lib/spark-examples-1.6.3-hadoop2.6.0.jar # Yarn Custer模式运行

bin/yarn application -list # 查看YARN运行的应用bin/yarn application -kill ApplicationID # 杀死YARN运行的应用

bin/spark-shell --master spark://centos1:7077 --deploy-mode client --driver-memory 512m --executor-memory 512m --num-executors 1 --executor-cores 1 # Standalone Client模式运行bin/spark-shell --master yarn --deploy-mode client --driver-memory 512m --executor-memory 512m --num-executors 1 --executor-cores 1 # Yarn Client模式运行

bin/spark-shell --help

Usage: ./bin/spark-shell [options]Options:  --master MASTER_URL         spark://host:port, mesos://host:port, yarn, or local.  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or                              on one of the worker machines inside the cluster ("cluster")                              (Default: client).  --class CLASS_NAME          Your application‘s main class (for Java / Scala apps).  --name NAME                 A name of your application.  --jars JARS                 Comma-separated list of local jars to include on the driver                              and executor classpaths.  --packages                  Comma-separated list of maven coordinates of jars to include                              on the driver and executor classpaths. Will search the local                              maven repo, then maven central and any additional remote                              repositories given by --repositories. The format for the                              coordinates should be groupId:artifactId:version.  --exclude-packages          Comma-separated list of groupId:artifactId, to exclude while                              resolving the dependencies provided in --packages to avoid                              dependency conflicts.  --repositories              Comma-separated list of additional remote repositories to                              search for the maven coordinates given with --packages.  --py-files PY_FILES         Comma-separated list of .zip, .egg, or .py files to place                              on the PYTHONPATH for Python apps.  --files FILES               Comma-separated list of files to be placed in the working                              directory of each executor.  --conf PROP=VALUE           Arbitrary Spark configuration property.  --properties-file FILE      Path to a file from which to load extra properties. If not                              specified, this will look for conf/spark-defaults.conf.  --driver-memory MEM         Memory for driver (e.g. 1000M, 2G) (Default: 1024M).  --driver-java-options       Extra Java options to pass to the driver.  --driver-library-path       Extra library path entries to pass to the driver.  --driver-class-path         Extra class path entries to pass to the driver. Note that                              jars added with --jars are automatically included in the                              classpath.  --executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).  --proxy-user NAME           User to impersonate when submitting the application.  --help, -h                  Show this help message and exit  --verbose, -v               Print additional debug output  --version,                  Print the version of current Spark Spark standalone with cluster deploy mode only:  --driver-cores NUM          Cores for driver (Default: 1). Spark standalone or Mesos with cluster deploy mode only:  --supervise                 If given, restarts the driver on failure.  --kill SUBMISSION_ID        If given, kills the driver specified.  --status SUBMISSION_ID      If given, requests the status of the driver specified. Spark standalone and Mesos only:  --total-executor-cores NUM  Total cores for all executors. Spark standalone and YARN only:  --executor-cores NUM        Number of cores per executor. (Default: 1 in YARN mode,                              or all available cores on the worker in standalone mode) YARN-only:  --driver-cores NUM          Number of cores used by the driver, only in cluster mode                              (Default: 1).  --queue QUEUE_NAME          The YARN queue to submit to (Default: "default").  --num-executors NUM         Number of executors to launch (Default: 2).  --archives ARCHIVES         Comma separated list of archives to be extracted into the                              working directory of each executor.  --principal PRINCIPAL       Principal to be used to login to KDC, while running on                              secure HDFS.  --keytab KEYTAB             The full path to the file that contains the keytab for the                              principal specified above. This keytab will be copied to                              the node running the Application Master via the Secure                              Distributed Cache, for renewing the login tickets and the                              delegation tokens periodically.

bin/spark-submit --help

Usage: spark-submit [options] <app jar | python file> [app arguments]Usage: spark-submit --kill [submission ID] --master [spark://...]Usage: spark-submit --status [submission ID] --master [spark://...]Options:  --master MASTER_URL         spark://host:port, mesos://host:port, yarn, or local.  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or                              on one of the worker machines inside the cluster ("cluster")                              (Default: client).  --class CLASS_NAME          Your application‘s main class (for Java / Scala apps).  --name NAME                 A name of your application.  --jars JARS                 Comma-separated list of local jars to include on the driver                              and executor classpaths.  --packages                  Comma-separated list of maven coordinates of jars to include                              on the driver and executor classpaths. Will search the local                              maven repo, then maven central and any additional remote                              repositories given by --repositories. The format for the                              coordinates should be groupId:artifactId:version.  --exclude-packages          Comma-separated list of groupId:artifactId, to exclude while                              resolving the dependencies provided in --packages to avoid                              dependency conflicts.  --repositories              Comma-separated list of additional remote repositories to                              search for the maven coordinates given with --packages.  --py-files PY_FILES         Comma-separated list of .zip, .egg, or .py files to place                              on the PYTHONPATH for Python apps.  --files FILES               Comma-separated list of files to be placed in the working                              directory of each executor.  --conf PROP=VALUE           Arbitrary Spark configuration property.  --properties-file FILE      Path to a file from which to load extra properties. If not                              specified, this will look for conf/spark-defaults.conf.  --driver-memory MEM         Memory for driver (e.g. 1000M, 2G) (Default: 1024M).  --driver-java-options       Extra Java options to pass to the driver.  --driver-library-path       Extra library path entries to pass to the driver.  --driver-class-path         Extra class path entries to pass to the driver. Note that                              jars added with --jars are automatically included in the                              classpath.  --executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).  --proxy-user NAME           User to impersonate when submitting the application.  --help, -h                  Show this help message and exit  --verbose, -v               Print additional debug output  --version,                  Print the version of current Spark Spark standalone with cluster deploy mode only:  --driver-cores NUM          Cores for driver (Default: 1). Spark standalone or Mesos with cluster deploy mode only:  --supervise                 If given, restarts the driver on failure.  --kill SUBMISSION_ID        If given, kills the driver specified.  --status SUBMISSION_ID      If given, requests the status of the driver specified. Spark standalone and Mesos only:  --total-executor-cores NUM  Total cores for all executors. Spark standalone and YARN only:  --executor-cores NUM        Number of cores per executor. (Default: 1 in YARN mode,                              or all available cores on the worker in standalone mode) YARN-only:  --driver-cores NUM          Number of cores used by the driver, only in cluster mode                              (Default: 1).  --queue QUEUE_NAME          The YARN queue to submit to (Default: "default").  --num-executors NUM         Number of executors to launch (Default: 2).  --archives ARCHIVES         Comma separated list of archives to be extracted into the                              working directory of each executor.  --principal PRINCIPAL       Principal to be used to login to KDC, while running on                              secure HDFS.  --keytab KEYTAB             The full path to the file that contains the keytab for the                              principal specified above. This keytab will be copied to                              the node running the Application Master via the Secure                              Distributed Cache, for renewing the login tickets and the                              delegation tokens periodically.

import org.apache.spark.SparkConfimport org.apache.spark.SparkContextimport org.apache.spark.sql.SQLContextimport org.apache.spark.sql.hive.HiveContextobject Test {  def main(args: Array[String]): Unit = {    val conf = new SparkConf().setAppName("Test")    val sc = new SparkContext(conf)        // ...  }}

bin/spark-submit --master spark://ubuntu1:7077 --class org.apache.spark.examples.SparkPi lib/spark-examples-1.6.3-hadoop2.6.0.jar

 1 package org.apache.spark.rdd 2  3 // … 4  5 /** 6  * A Resilient Distributed Dataset (RDD), the basic abstraction in Spark. Represents an immutable, 7  * partitioned collection of elements that can be operated on in parallel. This class contains the 8  * basic operations available on all RDDs, such as `map`, `filter`, and `persist`. In addition, 9  * [[org.apache.spark.rdd.PairRDDFunctions]] contains operations available only on RDDs of key-value10  * pairs, such as `groupByKey` and `join`;11  * [[org.apache.spark.rdd.DoubleRDDFunctions]] contains operations available only on RDDs of12  * Doubles; and13  * [[org.apache.spark.rdd.SequenceFileRDDFunctions]] contains operations available on RDDs that14  * can be saved as SequenceFiles.15  * All operations are automatically available on any RDD of the right type (e.g. RDD[(Int, Int)]16  * through implicit.17  *18  * Internally, each RDD is characterized by five main properties:19  *20  *  - A list of partitions21  *  - A function for computing each split22  *  - A list of dependencies on other RDDs23  *  - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)24  *  - Optionally, a list of preferred locations to compute each split on (e.g. block locations for25  *    an HDFS file)26  *27  * All of the scheduling and execution in Spark is done based on these methods, allowing each RDD28  * to implement its own way of computing itself. Indeed, users can implement custom RDDs (e.g. for29  * reading data from a new storage system) by overriding these functions. Please refer to the30  * [[http://www.cs.berkeley.edu/~matei/papers/2012/nsdi_spark.pdf Spark paper]] for more details31  * on RDD internals.32  */33 abstract class RDD[T: ClassTag](34     @transient private var _sc: SparkContext,35     @transient private var deps: Seq[Dependency[_]]36   ) extends Serializable with Logging {37   // ...38   39   // =======================================================================40   // Methods that should be implemented by subclasses of RDD41   // =======================================================================42 43   /**44    * :: DeveloperApi ::45    * Implemented by subclasses to compute a given partition.46    */47   @DeveloperApi48   def compute(split: Partition, context: TaskContext): Iterator[T]49 50   /**51    * Implemented by subclasses to return the set of partitions in this RDD. This method will only52    * be called once, so it is safe to implement a time-consuming computation in it.53    */54   protected def getPartitions: Array[Partition]55 56   /**57    * Implemented by subclasses to return how this RDD depends on parent RDDs. This method will only58    * be called once, so it is safe to implement a time-consuming computation in it.59    */60   protected def getDependencies: Seq[Dependency[_]] = deps61 62   /**63    * Optionally overridden by subclasses to specify placement preferences.64    */65   protected def getPreferredLocations(split: Partition): Seq[String] = Nil66 67   /** Optionally overridden by subclasses to specify how they are partitioned. */68   @transient val partitioner: Option[Partitioner] = None69 70   // ...71 }

hadoop fs -mkdir -p /test/wordcounthadoop fs -put README.md /test/wordcount

spark-shell --master spark://centos1:7077

1 import org.apache.log4j.{Logger,Level}2 Logger.getLogger("org.apache.spark").setLevel(Level.WARN)3 Logger.getLogger("org.apache.spark.sql").setLevel(Level.WARN)4 val textFile = sc.textFile("/test/wordcount/README.md")5 val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((count1, count2) => count1 + count2)6 wordCounts.saveAsTextFile("/test/wordcount/result")7 wordCounts.collect

hadoop fs -cat /test/wordcount/WordCounts

def textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String]

def parallelize[T: ClassTag](seq: Seq[T], numSlices: Int = defaultParallelism): RDD[T]

1 val testRdd = sc.parallelize(List("A", "B", "C", "D"), 2)2 testRdd.partitions.size // 分区数23 testRdd.toDebugString // 查看Lineage

def persist(newLevel: StorageLevel): this.type

def cache()

1 val file = sc.textFile("/test/wordcount/README.md") // 可分别尝试小文件和超大文件（视内存）2 file.cache // 缓存到内存，lazy操作3 file.count // Action操作触发lazy4 file.unpersit // 释放缓存，eager操作

1 val textFile = sc.textFile("/test/README.md")2 textFile.toDebugString // 查看Lineage3 textFile.partitions.size // 分区数44 sc.defaultParallelism // 默认并行数4

 1 // … 2 package org.apache.spark 3 // … 4 object Partitioner { 5   /** 6    * Choose a partitioner to use for a cogroup-like operation between a number of RDDs. 7    * 8    * If any of the RDDs already has a partitioner, choose that one. 9    *10    * Otherwise, we use a default HashPartitioner. For the number of partitions, if11    * spark.default.parallelism is set, then we‘ll use the value from SparkContext12    * defaultParallelism, otherwise we‘ll use the max number of upstream partitions.13    *14    * Unless spark.default.parallelism is set, the number of partitions will be the15    * same as the number of partitions in the largest upstream RDD, as this should16    * be least likely to cause out-of-memory errors.17    *18    * We use two method parameters (rdd, others) to enforce callers passing at least 1 RDD.19    */20   def defaultPartitioner(rdd: RDD[_], others: RDD[_]*): Partitioner = {21     val bySize = (Seq(rdd) ++ others).sortBy(_.partitions.size).reverse22     for (r <- bySize if r.partitioner.isDefined && r.partitioner.get.numPartitions > 0) {23       return r.partitioner.get24     }25     if (rdd.context.conf.contains("spark.default.parallelism")) {26       new HashPartitioner(rdd.context.defaultParallelism)27     } else {28       new HashPartitioner(bySize.head.partitions.size)29     }30   }31 }32 // …

1 val broadcastVar = sc.broadcast(Array(1, 2, 3)) // 初始化2 broadcastVar.value // 获取

1 val accumulator = sc.accumulator(0) // 初始化2 accumulator += 100 // 累加3 accumulator.value // 获取

def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null): RDD[T]

1 val rdd = sc.parallelize(List("A", "B", "C", "D", "E"), 2)2 rdd.partitions.size // 分区数23 rdd.coalesce(1).partitions.size // 分区数2→14 rdd.coalesce(4).partitions.size // 分区数2→4无效，仍然是25 rdd.coalesce(4, true).partitions.size // 分区数2→4

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

 1 /** 2 * Return a new RDD that has exactly numPartitions partitions. 3 * 4 * Can increase or decrease the level of parallelism in this RDD. Internally, this uses 5 * a shuffle to redistribute data. 6 * 7 * If you are decreasing the number of partitions in this RDD, consider using `coalesce`, 8 * which can avoid performing a shuffle. 9 */10 def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {11   coalesce(numPartitions, shuffle = true)12 }

1 val rdd = sc.parallelize(List("A", "B", "C", "D", "E"), 2)2 rdd.partitions.size // 分区数23 rdd.repartition(1).partitions.size // 分区数2→14 rdd.repartition(4).partitions.size // 分区数2→4

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U

1 var rdd = sc.makeRDD(1 to 10, 2)2 rdd.aggregate(2)({(x: Int, y: Int) => x + y}, {(a: Int, b: Int) => a * b})3 // 分区1：2 + 1 + 2 + 3 + 4 + 5 = 17，4 // 分区2：2 + 6 + 7 + 8 + 9 + 10 = 425 // 最后：2 * 17 * 42 = 1428

def reduce(f: (T, T) => T): T

1 var rdd = sc.makeRDD(1 to 10, 2)2 rdd.reduce((x, y) => if (x > y) x else y) // 求最大值，103 import java.lang.Math4 rdd.reduce((x, y) => Math.max(x, y)) // 求最大值，与上面逻辑一致5 rdd.reduce((x, y) => x + y) // 求和，55

def fold(zeroValue: T)(op: (T, T) => T): T

1 var rdd = sc.makeRDD(1 to 10, 2)2 rdd.fold(100)((x, y) => if (x > y) x else y) // 求最大值，1003 import java.lang.Math4 rdd.reduce((x, y) => Math.max(x, y)) // 求最大值，与上面逻辑一致5 rdd.reduce((x, y) => x + y) // 求和，155

def cartesian[U: ClassTag](other: RDD[U]): RDD[(T, U)]

1 val rdd1 = sc.parallelize(List[String]("A", "B", "C", "D", "E"), 2)2 val rdd2 = sc.parallelize(List[Int](1, 2, 3, 4, 5, 6), 2)3 val rdd3 = rdd1 cartesian rdd24 rdd3.collect5 val rdd4 = rdd2 cartesian rdd16 rdd4.collect

def union(other: RDD[T]): RDD[T]

1 val rdd1 = sc.parallelize(List[String]("A", "B", "C"), 2)2 val rdd2 = sc.parallelize(List[String]("D", "E"), 1)3 rdd1.partitions.size // 分区数24 rdd2.partitions.size // 分区数15 val rdd3 = rdd1 union rdd26 rdd3.partitions.size // 分区数37 rdd3.collect // Array(A, B, C, C, D)

def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]

1 val rdd1 = sc.parallelize(1 to 4, 2)2 val rdd2 = sc.parallelize("a b c d".split(" "), 2)3 val rdd3 = rdd1 zip rdd24 rdd3.collect // Array((1, a), (2, b), (3, c), (4, d))

def zipPartitions[B: ClassTag, V: ClassTag](rdd2: RDD[B], preservesPartitioning: Boolean)(f: (Iterator[T], Iterator[B]) => Iterator[V]): RDD[V]def zipPartitions[B: ClassTag, V: ClassTag](rdd2: RDD[B])(f: (Iterator[T], Iterator[B]) => Iterator[V]): RDD[V]def zipPartitions[B: ClassTag, C: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C], preservesPartitioning: Boolean)(f: (Iterator[T], Iterator[B], Iterator[C]) => Iterator[V]): RDD[V]def zipPartitions[B: ClassTag, C: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C])(f: (Iterator[T], Iterator[B], Iterator[C]) => Iterator[V]): RDD[V]def zipPartitions[B: ClassTag, C: ClassTag, D: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C], rdd4: RDD[D], preservesPartitioning: Boolean)(f: (Iterator[T], Iterator[B], Iterator[C], Iterator[D]) => Iterator[V]): RDD[V]def zipPartitions[B: ClassTag, C: ClassTag, D: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C], rdd4: RDD[D])(f: (Iterator[T], Iterator[B], Iterator[C], Iterator[D]) => Iterator[V]): RDD[V]

val rdd1 = sc.parallelize(1 to 4, 2)val rdd2 = sc.parallelize("a b c d".split(" "), 2)def zipFunc(aIter: Iterator[Int], bIter: Iterator[String]): Iterator[String] = {  var list = List[String]()  while (aIter.hasNext && bIter.hasNext) {    val str = aIter.next + "+" + bIter.next    list ::= str  }  list.iterator}val rdd3 = rdd1.zipPartitions(rdd1, rdd2)(zipFunc)rdd3.collect // Array(1+a, 2+b, 3+c, 4+d)

def zipWithUniqueId(): RDD[(T, Long)]

1 val rdd = sc.parallelize("a b c d".split(" "), 2)2 rdd.zipWithUniqueId.collect // Array((a, 0), (b, 1), (c, 2), (d, 3))

def histogram(buckets: Array[Double], evenBuckets: Boolean = false): Array[Long]def histogram(bucketCount: Int): Pair[Array[Double], Array[Long]]

1 val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)2 rdd.histogram(Array(0.0, 4.1, 9.0))3 rdd.histogram(Array(0.0, 4.1, 9.0), true)4 rdd.histogram(3)

def mean(): Doubledef meanApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]

1 val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)2 rdd.mean

def sampleStdev(): Double

1 val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)2 rdd.sampleStdev

def sampleVariance(): Double

val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)rdd.sampleVariance

def stats(): StatCounter

val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)rdd.stats

def stdev(): Double

1 val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)2 rdd.stdev

def sum(): Doubledef sumApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]

1 val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)2 rdd.sum3 rdd.sumApprox

def variance(): Double

1 val rdd = sc.parallelize(List(1.1, 1.2, 2.1, 2.2, 2, 3, 4.1, 4.3, 7.1, 8.3, 9.3), 2)2 rdd.variance

def aggregateByKey[U: ClassTag](zeroValue: U)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]def aggregateByKey[U: ClassTag](zeroValue: U, numPartitions: Int)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]def aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) => U, combOp: (U, U) => U): RDD[(K, U)]

1 val rdd = sc.parallelize(List((1, 3), (1, 2), (1, 4), (2, 3))) // Array((1,9), (2,3))2 import java.lang.Math3 rdd.aggregateByKey(1)({(x: Int, y: Int) => Math.max(x, y)}, {(a: Int, b: Int) => a + b}).collect

def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]

1 val rdd = sc.parallelize(List((1, "www"), (1, "iteblog"), (1, "com"), (2, "bbs"), (2, "iteblog"), (2, "com"), (3, "good")))2 rdd.combineByKey(List(_), (x: List[String], y: String) => y :: x, (x: List[String], y: List[String]) => x ::: y).collect // Array((1, List(www, iteblog, com)), (2, List(bbs, iteblog, com)), (3, List(good)))

1 val rdd = sc.parallelize(List(("iteblog", 1), ("bbs", 1), ("iteblog", 3)))2 rdd.combineByKey(x => x, (x: Int, y: Int) => x + y, (x: Int, y: Int) => x + y).collect // Array((iteblog, 4), (bbs, 1))

def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]def foldByKey(zeroValue: V, numPartitions: Int)(func: (V, V) => V): RDD[(K, V)]def foldByKey(zeroValue: V, partitioner: Partitioner)(func: (V, V) => V): RDD[(K, V)]

1 var rdd = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))2 rdd.foldByKey(0)(_+_).collect // Array((A,2), (B,3), (C,1))3 rdd.foldByKey(2)(_+_).collect // Array((A,6), (B,7), (C,3))

def reduceByKey(func: (V, V) => V): RDD[(K, V)]def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]

1 var rdd = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",1)))2 rdd.reduceByKey(_+_).collect // Array((A,2), (B,3), (C,1))

def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]def leftOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, Option[W]))]def leftOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, Option[W]))]def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], W))]def rightOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Option[V], W))]def rightOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], W))]def fullOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], Option[W]))]def fullOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Option[V], Option[W]))]def fullOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], Option[W]))]

1 val rdd1 = sc.parallelize(List(("Tom", 21), ("Jerry", 31), ("Mary", 23)))2 val rdd2 = sc.parallelize(List(("Tom", ‘m‘), ("Mary", ‘f‘), ("Henry", ‘m‘)))3 val rdd3 = rdd1 join rdd24 rdd3.collect // Array((Mary, (23, f)), (Tom, (21, m)))

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]

1 val rdd1 = sc.parallelize(List((3, "a"), (7, "b"), (5, "c"), (3, "b"), (6, "c"), (9, "d")), 3)2 val rdd2 = rdd1.sortByKey3 rdd2.collect // Array((3, a), (3, b), (5, c), (6, c), (7, b), (9, d))4 rdd2.partitions.size // 分区数35 val rdd3 = rdd1.sortByKey(true, 2)6 rdd3.collect7 rdd3.partitions.size // 分区数2

def repartitionAndSortWithinPartitions(partitioner: Partitioner): RDD[(K, V)]

1 val rdd1 = sc.parallelize(List((3, "a"), (7, "b"), (5, "c"), (3, "b"), (6, "c"), (9, "d")), 3)2 import org.apache.spark.HashPartitioner3 val rdd2 = rdd1.repartitionAndSortWithinPartitions(new HashPartitioner(2)) // HashPartitioner(2)以key对分区数取模分区，所以奇数、偶数分到两个分区。4 rdd2.collect // Array((6, c), (3, a), (3, b), (5, c), (7, b), (9, d))5 rdd1.partitions.size // 分区数36 rdd2.partitions.size // 分区数2

1,1,460028714280218,360,2015-05-01,7,11161,2,460028714280219,qq,2015-05-02,8,1211,3,460028714280220,yy,2015-05-03,9,1221,4,460028714280221,360,2015-05-04,10,1192,1,460028714280222,yy,2015-05-05,5,11192,2,460028714280223,360,2015-05-01,12,1212,3,460028714280224,qq,2015-05-02,13,1223,1,460028714280225,qq,2015-05-03,1,11173,2,460028714280226,qq,2015-05-04,9,11183,3,460028714280227,qq,2015-05-05,10,1201,1,460028714280218,360,2015-06-01,7,11181,2,460028714280219,qq,2015-06-02,8,11191,3,460028714280220,yy,2015-06-03,9,11201,4,460028714280221,360,2015-06-04,10,1192,1,460028714280222,yy,2015-06-05,11,11182,2,460028714280223,360,2015-06-02,4,1212,3,460028714280224,qq,2015-06-03,17,11193,1,460028714280225,qq,2015-06-04,18,1193,2,460028714280226,qq,2015-06-05,19,11193,3,460028714280227,qq,2015-06-10,120,121

hadoop fs -put mobile.csv /test/

1 val fields = sc.broadcast(List("nodeid", "ci", "imei", "app", "time", "uplinkbytes", "downlinkbytes"))2 val mobile = sc.textFile("/test/mobile.csv").map(_.split(",")).filter(line => line.length != fields.value.length)

1 mobile.map(line => (line(fields.value.indexOf("app")), 1)).reduceByKey(_+_).map(appCount => (appCount._2, appCount._1)).sortByKey(false).map(appCount => (appCount._1, appCount._2)).repartition(1).saveAsTextFile("/text/result.csv") // Array((qq, 10), (360, 6), (yy, 4))

1 mobile.map(line => line(fields.value.indexOf("imei")) + ":" + line(fields.value.indexOf("time"))).distinct().map(imeiTime => (imeiTime.split(":")(1), 1)).reduceByKey(_+_).sortByKey().collect // Array((2015-05-01, 2), (2015-05-02, 2), (2015-05-03, 2), (2015-05-04, 2), (2015-05-05, 2), (2015-06-01, 2), (2015-06-03, 2), (2015-06-04, 2), (2015-06-05, 2), (2015-06-10, 1))

1 mobile.map { line =>2   val time = line(fields.value.indexOf("time"))3   val month = time.substring(0, time.lastIndexOf("-"))4   line(fields.value.indexOf("imei")) + ":" + month5 }.distinct.map { imeiMonth => (imeiMonth.split(":")(1), 1) }.reduceByKey(_+_).sortByKey().collect // Array((2015-05, 10), (2015-06, 10))

1 mobile.map { line =>2   val uplinkbytes = line(fields.value.indexOf("uplinkbytes"))3   val downlinkbytes = line(fields.value.indexOf("downlinkbytes"))4   (line(fields.value.indexOf("app")), (uplinkbytes, downlinkbytes))5 }.reduceByKey((updownlinkbytes1, updownlinkbytes2) => (updownlinkbytes1._1 + updownlinkbytes2._1, updownlinkbytes1._2 + updownlinkbytes2._2)).collect // Array((yy, (34.0, 3479.0)), (qq, (117.0, 6195.0)), (360, (54.0, 2714.0)))