データアルゴリズムHadoop/Sparkビッグデータ処理---第五章
16177 ワード
この章では、反転ソート(5つの方法)について説明します.従来のMapReduce 従来のspark spark SQLの方法 従来のScala法 Scalaのspark SQLメソッド 本章で取り扱うべき問題
1つの単語が隣接する前後の2桁の数を統計し、w 1、w 2、w 3、w 4、w 5、w 6があれば:
単語
領域(+-2)
W1
W2,W3
W2
W1,W3,W4
W3
W1,W2,W4,W5
W4
W2,W3,W5,W6
W5
W3,W4,W6
W6
W4,W5
最終的に出力するのは(word,neighbor,frequency)
従来のMapReduce
クラス名
クラスの説明
RelativeFrequencyDriver
ジョブドライブの発行
RelativeFrequencyMapper
map()関数
RelativeFrequencyReducer
reduce()関数
RelativeFrequencyCombiner
combine-map()転送を減らす
OrderInversionPartitioner
キーに従ってパーティション化
PairOfWords
表示語対(Word 1,Word 2)
従来のspark
Spark SQLの方法
従来のScala法
Scalaのspark SQLメソッド
1つの単語が隣接する前後の2桁の数を統計し、w 1、w 2、w 3、w 4、w 5、w 6があれば:
単語
領域(+-2)
W1
W2,W3
W2
W1,W3,W4
W3
W1,W2,W4,W5
W4
W2,W3,W5,W6
W5
W3,W4,W6
W6
W4,W5
最終的に出力するのは(word,neighbor,frequency)
従来のMapReduce
クラス名
クラスの説明
RelativeFrequencyDriver
ジョブドライブの発行
RelativeFrequencyMapper
map()関数
RelativeFrequencyReducer
reduce()関数
RelativeFrequencyCombiner
combine-map()転送を減らす
OrderInversionPartitioner
キーに従ってパーティション化
PairOfWords
表示語対(Word 1,Word 2)
//map
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String[] tokens = StringUtils.split(value.toString(), " ");
//String[] tokens = StringUtils.split(value.toString(), "\\s+");
if ((tokens == null) || (tokens.length < 2)) {
return;
}
//
for (int i = 0; i < tokens.length; i++) {
tokens[i] = tokens[i].replaceAll("\\W+", "");
if (tokens[i].equals("")) {
continue;
}
pair.setWord(tokens[i]);
int start = (i - neighborWindow < 0) ? 0 : i - neighborWindow;
int end = (i + neighborWindow >= tokens.length) ? tokens.length - 1 : i + neighborWindow;
for (int j = start; j <= end; j++) {
if (j == i) {
continue;
}
pair.setNeighbor(tokens[j].replaceAll("\\W", ""));
context.write(pair, ONE);
}
//
pair.setNeighbor("*");
totalCount.set(end - start);
context.write(pair, totalCount);
}
}
//reduce
@Override
protected void reduce(PairOfWords key, Iterable values, Context context)
throws IOException, InterruptedException {
// * ， count totalCount
if (key.getNeighbor().equals("*")) {
if (key.getWord().equals(currentWord)) {
totalCount += totalCount + getTotalCount(values);
} else {
currentWord = key.getWord();
totalCount = getTotalCount(values);
}
} else {
// word， getTotalCount
int count = getTotalCount(values);
relativeCount.set((double) count / totalCount);
context.write(key, relativeCount);
}
}
従来のspark
public static void main(String[] args) {
if (args.length < 3) {
System.out.println("Usage: RelativeFrequencyJava ");
System.exit(1);
}
SparkConf sparkConf = new SparkConf().setAppName("RelativeFrequency");
JavaSparkContext sc = new JavaSparkContext(sparkConf);
int neighborWindow = Integer.parseInt(args[0]);
String input = args[1];
String output = args[2];
final Broadcast brodcastWindow = sc.broadcast(neighborWindow);
JavaRDD rawData = sc.textFile(input);
/*
* Transform the input to the format: (word, (neighbour, 1))
*/
JavaPairRDD> pairs = rawData.flatMapToPair(
new PairFlatMapFunction>() {
private static final long serialVersionUID = -6098905144106374491L;
@Override
public java.util.Iterator>> call(String line) throws Exception {
List>> list = new ArrayList>>();
String[] tokens = line.split("\\s");
for (int i = 0; i < tokens.length; i++) {
int start = (i - brodcastWindow.value() < 0) ? 0 : i - brodcastWindow.value();
int end = (i + brodcastWindow.value() >= tokens.length) ? tokens.length - 1 : i + brodcastWindow.value();
for (int j = start; j <= end; j++) {
if (j != i) {
list.add(new Tuple2>(tokens[i], new Tuple2(tokens[j], 1)));
} else {
// do nothing
continue;
}
}
}
return list.iterator();
}
}
);
// (word, sum(word))
//PairFunction T => Tuple2
JavaPairRDD totalByKey = pairs.mapToPair(
new PairFunction>, String, Integer>() {
private static final long serialVersionUID = -213550053743494205L;
@Override
public Tuple2 call(Tuple2> tuple) throws Exception {
return new Tuple2(tuple._1, tuple._2._2);
}
}).reduceByKey(
new Function2() {
private static final long serialVersionUID = -2380022035302195793L;
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return (v1 + v2);
}
});
JavaPairRDD>> grouped = pairs.groupByKey();
// (word, (neighbour, 1)) -> (word, (neighbour, sum(neighbour)))
//flatMapValues value ， key
JavaPairRDD> uniquePairs = grouped.flatMapValues(
//Function -> R call(T1 v1)
new Function>, Iterable>>() {
private static final long serialVersionUID = 5790208031487657081L;
@Override
public Iterable> call(Iterable> values) throws Exception {
Map map = new HashMap<>();
List> list = new ArrayList<>();
Iterator> iterator = values.iterator();
while (iterator.hasNext()) {
Tuple2 value = iterator.next();
int total = value._2;
if (map.containsKey(value._1)) {
total += map.get(value._1);
}
map.put(value._1, total);
}
for (Map.Entry kv : map.entrySet()) {
list.add(new Tuple2(kv.getKey(), kv.getValue()));
}
return list;
}
});
// (word, ((neighbour, sum(neighbour)), sum(word)))
JavaPairRDD, Integer>> joined = uniquePairs.join(totalByKey);
// ((key, neighbour), sum(neighbour)/sum(word))
JavaPairRDD, Double> relativeFrequency = joined.mapToPair(
new PairFunction, Integer>>, Tuple2, Double>() {
private static final long serialVersionUID = 3870784537024717320L;
@Override
public Tuple2, Double> call(Tuple2, Integer>> tuple) throws Exception {
return new Tuple2, Double>(new Tuple2(tuple._1, tuple._2._1._1), ((double) tuple._2._1._2 / tuple._2._2));
}
});
// For saving the output in tab separated format
// ((key, neighbour), relative_frequency)
// String
JavaRDD formatResult_tab_separated = relativeFrequency.map(
new Function, Double>, String>() {
private static final long serialVersionUID = 7312542139027147922L;
@Override
public String call(Tuple2, Double> tuple) throws Exception {
return tuple._1._1 + "\t" + tuple._1._2 + "\t" + tuple._2;
}
});
// save output
formatResult_tab_separated.saveAsTextFile(output);
// done
sc.close();
}
Spark SQLの方法
public static void main(String[] args) {
if (args.length < 3) {
System.out.println("Usage: SparkSQLRelativeFrequency ");
System.exit(1);
}
SparkConf sparkConf = new SparkConf().setAppName("SparkSQLRelativeFrequency");
// SparkSQL SparkSession
SparkSession spark = SparkSession
.builder()
.appName("SparkSQLRelativeFrequency")
.config(sparkConf)
.getOrCreate();
JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
int neighborWindow = Integer.parseInt(args[0]);
String input = args[1];
String output = args[2];
final Broadcast brodcastWindow = sc.broadcast(neighborWindow);
/*
* Schema ， frequency
* Schema (word, neighbour, frequency)
*/
StructType rfSchema = new StructType(new StructField[]{
new StructField("word", DataTypes.StringType, false, Metadata.empty()),
new StructField("neighbour", DataTypes.StringType, false, Metadata.empty()),
new StructField("frequency", DataTypes.IntegerType, false, Metadata.empty())});
JavaRDD rawData = sc.textFile(input);
/*
* Transform the input to the format: (word, (neighbour, 1))
*/
JavaRDD rowRDD = rawData
.flatMap(new FlatMapFunction() {
private static final long serialVersionUID = 5481855142090322683L;
@Override
public Iterator call(String line) throws Exception {
List list = new ArrayList<>();
String[] tokens = line.split("\\s");
for (int i = 0; i < tokens.length; i++) {
int start = (i - brodcastWindow.value() < 0) ? 0
: i - brodcastWindow.value();
int end = (i + brodcastWindow.value() >= tokens.length) ? tokens.length - 1
: i + brodcastWindow.value();
for (int j = start; j <= end; j++) {
if (j != i) {
list.add(RowFactory.create(tokens[i], tokens[j], 1));
} else {
// do nothing
continue;
}
}
}
return list.iterator();
}
});
// DataFrame
Dataset rfDataset = spark.createDataFrame(rowRDD, rfSchema);
// rfDataset table，
rfDataset.createOrReplaceTempView("rfTable");
String query = "SELECT a.word, a.neighbour, (a.feq_total/b.total) rf "
+ "FROM (SELECT word, neighbour, SUM(frequency) feq_total FROM rfTable GROUP BY word, neighbour) a "
+ "INNER JOIN (SELECT word, SUM(frequency) as total FROM rfTable GROUP BY word) b ON a.word = b.word";
Dataset sqlResult = spark.sql(query);
sqlResult.show(); // print first 20 records on the console
sqlResult.write().parquet(output + "/parquetFormat"); // saves output in compressed Parquet format, recommended for large projects.
sqlResult.rdd().saveAsTextFile(output + "/textFormat"); // to see output via cat command
// done
sc.close();
spark.stop();
}
従来のScala法
def main(args: Array[String]): Unit = {
if (args.size < 3) {
println("Usage: RelativeFrequency ")
sys.exit(1)
}
val sparkConf = new SparkConf().setAppName("RelativeFrequency")
val sc = new SparkContext(sparkConf)
val neighborWindow = args(0).toInt
val input = args(1)
val output = args(2)
val brodcastWindow = sc.broadcast(neighborWindow)
val rawData = sc.textFile(input)
/*
* Transform the input to the format:
* (word, (neighbour, 1))
*/
val pairs = rawData.flatMap(line => {
val tokens = line.split("\\s")
for {
i = tokens.length) tokens.length - 1 else i + brodcastWindow.value
j (t._1, t._2._2)).reduceByKey(_ + _)
val grouped = pairs.groupByKey()
// (word, (neighbour, sum(neighbour)))
val uniquePairs = grouped.flatMapValues(_.groupBy(_._1).mapValues(_.unzip._2.sum))
// join RDD
// (word, ((neighbour, sum(neighbour)), sum(word)))
val joined = uniquePairs join totalByKey
// ((key, neighbour), sum(neighbour)/sum(word))
val relativeFrequency = joined.map(t => {
((t._1, t._2._1._1), (t._2._1._2.toDouble / t._2._2.toDouble))
})
// For saving the output in tab separated format
// ((key, neighbour), relative_frequency)
val formatResult_tab_separated = relativeFrequency.map(t => t._1._1 + "\t" + t._1._2 + "\t" + t._2)
formatResult_tab_separated.saveAsTextFile(output)
// done
sc.stop()
}
Scalaのspark SQLメソッド
def main(args: Array[String]): Unit = {
if (args.size < 3) {
println("Usage: SparkSQLRelativeFrequency ")
sys.exit(1)
}
val sparkConf = new SparkConf().setAppName("SparkSQLRelativeFrequency")
val spark = SparkSession
.builder()
.config(sparkConf)
.getOrCreate()
val sc = spark.sparkContext
val neighborWindow = args(0).toInt
val input = args(1)
val output = args(2)
val brodcastWindow = sc.broadcast(neighborWindow)
val rawData = sc.textFile(input)
/*
* Schema
* (word, neighbour, frequency)
*/
val rfSchema = StructType(Seq(
StructField("word", StringType, false),
StructField("neighbour", StringType, false),
StructField("frequency", IntegerType, false)))
/*
* Transform the input to the format:
* Row(word, neighbour, 1)
*/
// StructType
val rowRDD = rawData.flatMap(line => {
val tokens = line.split("\\s")
for {
i = tokens.length) tokens.length - 1 else i + brodcastWindow.value
j