// 这里通过textFile()方法，针对外部文件创建了一个RDD，lines，但是实际上，程序执行到这里为止，spark.txt文件的数据是不会加载到内存中的。lines，只是代表了一个指向spark.txt文件的引用。
// 这里对lines RDD进行了map算子，获取了一个转换后的lineLengths RDD。但是这里连数据都没有，当然也不会做任何操作。lineLengths RDD也只是一个概念上的东西而已。
// 之列，执行了一个action操作，reduce。此时就会触发之前所有transformation操作的执行，Spark会将操作拆分成多个task到多个机器上并行执行，每个task会在本地执行map操作，并且进行本地的reduce聚合。最后会进行一个全局的reduce聚合，然后将结果返回给Driver程序。
package sparkcore.java;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
/**
* 统计每行出现的次数,即同一行在文件里出现的次数
*/
public class LineCount {
public static void main(String[] args) {
// 创建SparkConf
SparkConf conf = new SparkConf().setAppName("LineCount").setMaster("local");
// 创建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 创建初始RDD，lines，每个元素是一行文本
JavaRDD<String> lines = sc.textFile("test.txt");
// 对lines RDD执行mapToPair算子，将每一行映射为(line, 1)的这种key-value对的格式
// 然后后面才能统计每一行出现的次数
JavaPairRDD<String, Integer> pairs = lines.mapToPair(
new PairFunction<String, String, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Tuple2<String, Integer> call(String t) throws Exception {
return new Tuple2<String, Integer>(t, 1);
}
});
// 对pairs RDD执行reduceByKey算子，统计出每一行出现的总次数
JavaPairRDD<String, Integer> lineCounts = pairs.reduceByKey(
new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
// 执行一个action操作，foreach，打印出每一行出现的次数
lineCounts.foreach(new VoidFunction<Tuple2<String, Integer>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Tuple2<String, Integer> t) throws Exception {
System.out.println(t._1 + " ： " + t._2);
}
});
// 关闭JavaSparkContext
sc.close();
}
}
package sparkcore.scala
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object LineCount {
def main(args: Array[String]) {
val conf = new SparkConf()
.setAppName("LineCount")
.setMaster("local")
val sc = new SparkContext(conf);
val lines = sc.textFile("test.txt", 1)
val pairs = lines.map { (_, 1) }
val lineCounts = pairs.reduceByKey { _ + _ }
lineCounts.foreach(lineCount => println(lineCount._1 + " : " + lineCount._2 ))
}
} 