1.定义输入和输出为:
- 输入数据格式为：年，月，日，温度。格式为2012，01，01，05
- 输出数据的格式为：年-月,温度。 格式为：2012-01，3，30 ，35。
2.使用MapReduce来完成上述二次排序的需求
– 2.1使用到的类的介绍
- SecondarySortDrivcer 驱动器类，定义输入输出，并注册插件类
- SecondarySortMapper 定义map()函数
- SecondarySortReduce 定义reduce()函数
- DataTemperatureGroupingComparator 定义如何对键分组
- DataTemperaturePair 将日期和温度对定义为java对象
- DataTemperaturePartitioner 定义定制分区器
– 2.2 关于启动类SecondarySortDriver.java
//SecondarySortDriver.java
public static int submitJob(String[] args) throws Exception {
//inputDir和outputDir存入为HDFS路径的话，则会连接到HDFS
//String[] args = new String[2];
//args[0] = inputDir;
//args[1] = outputDir;
//ToolRunner.run 的方法中实现了tool中的run，上面的对其进行了重写
int returnStatus = ToolRunner.run(new SecondarySortDriver(), args);
return returnStatus;
}
//ToolRunner.run方法，其中tool.run方法是接口Tool的方法
public static int run(Configuration conf, Tool tool, String[] args) throws Exception {
if(conf == null) {
conf = new Configuration();
}
GenericOptionsParser parser = new GenericOptionsParser(conf, args);
tool.setConf(conf);
String[] toolArgs = parser.getRemainingArgs();
return tool.run(toolArgs);
}
//之后就重写的run方法了
@Override
public int run(String[] args) throws Exception {
//定义了一些job的提交方式
Configuration conf = getConf();
Job job = new Job(conf);
job.setJarByClass(SecondarySortDriver.class);
job.setJobName("SecondarySortDriver");
// args[0] = input directory
// args[1] = output directory
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//定义job的提交方式
job.setOutputKeyClass(DateTemperaturePair.class);
job.setOutputValueClass(Text.class);
job.setMapperClass(SecondarySortMapper.class);
job.setReducerClass(SecondarySortReducer.class);
job.setPartitionerClass(DateTemperaturePartitioner.class);
job.setGroupingComparatorClass(DateTemperatureGroupingComparator.class);
boolean status = job.waitForCompletion(true);
theLogger.info("run(): status="+status);
return status ? 0 : 1;
}
– 2.3 关于Map函数类SecondarySortMapper.java
private final Text theTemperature = new Text();
//DateTemperaturePair可以理解为DateTemperaturePair，day,temperature的实体类，提供比较方法
private final DateTemperaturePair pair = new DateTemperaturePair();
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] tokens = line.split(",");
// YYYY = tokens[0]
// MM = tokens[1]
// DD = tokens[2]
// temperature = tokens[3]
String yearMonth = tokens[0] + tokens[1];
String day = tokens[2];
int temperature = Integer.parseInt(tokens[3]);
pair.setYearMonth(yearMonth);
pair.setDay(day);
pair.setTemperature(temperature);
theTemperature.set(tokens[3]);
context.write(pair, theTemperature);
}
- 其中DateTemperaturePair的比较方法为：
//如果yearMonth相同的话，才进行排序
@Override
public int compareTo(DateTemperaturePair pair) {
int compareValue = this.yearMonth.compareTo(pair.getYearMonth());
if (compareValue == 0) {
compareValue = temperature.compareTo(pair.getTemperature());
}
//return compareValue; // to sort ascending
return -1*compareValue; // to sort descending
}
– 2.4 关于Map函数类SecondarySortReducer.java
//注意传入的参数DateTemperaturePair, Text, Text, Text
public class SecondarySortReducer
extends Reducer<DateTemperaturePair, Text, Text, Text> {
@Override
protected void reduce(DateTemperaturePair key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
StringBuilder builder = new StringBuilder();
for (Text value : values) {
builder.append(value.toString());
builder.append(",");
}
context.write(key.getYearMonth(), new Text(builder.toString()));
}
}
– 2.5 关于运行函数的脚本
#cat run.sh
export JAVA_HOME=/usr/java/jdk8
export BOOK_HOME=/home/mp/data-algorithms-book
export APP_JAR=$BOOK_HOME/dist/data_algorithms_book.jar
INPUT=/secondary_sort/input
OUTPUT=/secondary_sort/output
$HADOOP_HOME/bin/hadoop -fs -rmr $OUTPUT
PROG = org.dataalgorithms.chap01.mapreduce.SecondarySortDriver
$HADOOP_HOME/bin/hadoop jar $APP_JAR $PROG $INPUT $OUTPUT
3.使用spark来完成上述二次排序的需求
– 3.1 需求的定义
* Input:
*
* name, time, value
* x,2,9
* y,2,5
* x,1,3
* y,1,7
* y,3,1
* x,3,6
* z,1,4
* z,2,8
* z,3,7
* z,4,0
*
*Output: generate a time-series looking like this:
* x => [(1,3), (2,9), (3,6)]
* y => [(1,7), (2,5), (3,1)]
* z => [(1,4), (2,8), (3,7), (4,0)]
– 3.2 SecondarySort类总结构
public class SecondarySort{
public static void main(String[] args) throws Exception {
//步骤2：读取输入参数并验证
//步骤3：创建一个javasparkcontext对象(ctx)
//步骤4：使用ctx创建JavaRDD<String>
//步骤5：JavaRDD<String>创建键值对，其中键是{name},值是{time,value}对
//步骤6：验证步骤5，打印出来
//步骤7：按键{name}对JavaRDD<String>元素分组
//步骤8：验证步骤7，
//步骤9：对归约器值排序，将得到最终输出
//步骤10：验证步骤9，
ctx.close();
System.exit(0);
}
}
– 3.2.1 步骤2:读取输入参数
if (args.length < 2) {
System.err.println("Usage: SecondarySortUsingGroupByKey <input> <output>");
System.exit(1);
}
String inputPath = args[0];
System.out.println("inputPath=" + inputPath);
String outputPath = args[1];
System.out.println("outputPath=" + outputPath);
– 3.2.2 步骤3:连接到sparkMaster
// STEP-2: Connect to the Spark master by creating JavaSparkContext object
final JavaSparkContext ctx = SparkUtil.createJavaSparkContext("SecondarySorting");
– 3.2.3 步骤4:创建javaRDD
// input record format: <name><,><time><,><value>
JavaRDD<String> lines = ctx.textFile(inputPath, 1);
– 3.2.4 步骤5:从avaRDD中创建键值读，从 <name><,><time><,><value>转换成一个<name，Tuple(time,value)组合>
JavaPairRDD<String, Tuple2<Integer, Integer>> pairs = lines.mapToPair(new PairFunction<String, String, Tuple2<Integer, Integer>>() {
@Override
//重写PairFunction中的 Tuple2<K, V> call(T t)方法，此处s 为 T
public Tuple2<String, Tuple2<Integer, Integer>> call(String s) {
String[] tokens = s.split(","); // x,2,5
System.out.println(tokens[0] + "," + tokens[1] + "," + tokens[2]);
Tuple2<Integer, Integer> timevalue = new Tuple2<Integer, Integer>(Integer.parseInt(tokens[1]), Integer.parseInt(tokens[2]));
//转换为对应的返回值
return new Tuple2<String, Tuple2<Integer, Integer>>(tokens[0], timevalue);
}
});
– 3.2.5 步骤6:验证步骤五的输出结果
/**
* OUTPUT:
* X,2,9
* Y,2,5
* X,1,3
* ……
*/
List<Tuple2<String, Tuple2<Integer, Integer>>> output = pairs.collect();
for (Tuple2 t : output) {
Tuple2<Integer, Integer> timevalue = (Tuple2<Integer, Integer>) t._2;
System.out.println(t._1 + "," + timevalue._1 + "," + timevalue._2);
}
– 3.2.6 步骤7:通过groupByKey对key进行分组
// STEP-6: We group JavaPairRDD<> elements by the key ({name}).
JavaPairRDD<String, Iterable<Tuple2<Integer, Integer>>> groups = pairs.groupByKey();
– 3.2.7 步骤8:验证步骤七的输入结果
/**
* OUTPUT:
* Y
* 2,5
* 1,7
* 3.1
* ……
*/
List<Tuple2<String, Iterable<Tuple2<Integer, Integer>>>> output2 = groups.collect();
for (Tuple2<String, Iterable<Tuple2<Integer, Integer>>> t : output2) {
Iterable<Tuple2<Integer, Integer>> list = t._2;
//输入key
System.out.println(t._1);
for (Tuple2<Integer, Integer> t2 : list) {
//输入value Tuple中的值
System.out.println(t2._1 + "," + t2._2);
}
System.out.println("=====");
}
– 3.2.8 步骤9:用mapValues对value的第一位进行排序
//mapValues方法可以对value进行排序，但是不影响key的顺序
JavaPairRDD<String, Iterable<Tuple2<Integer, Integer>>> sorted = groups.mapValues(new Function<Iterable<Tuple2<Integer, Integer>>, // input
Iterable<Tuple2<Integer, Integer>> // output
>() {
@Override
public Iterable<Tuple2<Integer, Integer>> call(Iterable<Tuple2<Integer, Integer>> s) {
List<Tuple2<Integer, Integer>> newList = new ArrayList<Tuple2<Integer, Integer>>(iterableToList(s));
//SparkTupleComparator中继承了Comparator并重写了它的sort方法
Collections.sort(newList, SparkTupleComparator.INSTANCE);
return newList;
}
});
– 3.2.9 步骤10:构造结果的打印规则并保持到HDFS
/**
* OUTPUT:
* (z,[(1,4),(2,8),(3,7)])
* ……
*/
List<Tuple2<String, Iterable<Tuple2<Integer, Integer>>>> output3 = sorted.collect();
for (Tuple2<String, Iterable<Tuple2<Integer, Integer>>> t : output3) {
Iterable<Tuple2<Integer, Integer>> list = t._2;
System.out.println(t._1);
for (Tuple2<Integer, Integer> t2 : list) {
System.out.println(t2._1 + "," + t2._2);
}
System.out.println("=====");
}
sorted.saveAsTextFile(outputPath);
System.exit(0);
}
4.使用scala完成需求
def main(args: Array[String]): Unit = {
//
if (args.length != 3) {
println("Usage <number-of-partitions> <input-path> <output-path>")
sys.exit(1)
}
// val partitions = args(0).toInt
// val inputPath = args(1)
// val outputPath = args(2)
val partitions = 3
val inputPath = "C:\\Users\\Administrator\\Desktop\\Book Code\\input.txt"
val outputPath = " C:\\Users\\Administrator\\Desktop\\Book Code\\output.txt"
val config = new SparkConf
config.setAppName("SecondarySort")
val sc = new SparkContext(config)
val input = sc.textFile(inputPath)
//------------------------------------------------
// each input line/record has the following format:
// <id><,><time><,><value>
//-------------------------------------------------
val valueToKey = input.map(x => {
val line = x.split(",")
//返回(<year-day,temperature>,temperature)
((line(0) + "-" + line(1), line(2).toInt), line(2).toInt)
})
//隐式转换比较的规则
implicit def tupleOrderingDesc = new Ordering[Tuple2[String, Int]] {
override def compare(x: Tuple2[String, Int], y: Tuple2[String, Int]): Int = {
if (y._1.compare(x._1) == 0) y._2.compare(x._2)
else y._1.compare(x._1)
}
}
//排序之后的列表
val sorted = valueToKey.repartitionAndSortWithinPartitions(new CustomPartitioner(partitions))
//进行结果的转换
val result = sorted.map {
case (k, v) => (k._1, v)
}
//保持到文件
result.saveAsTextFile(outputPath)
// done
sc.stop()
}
