Repository navigation
Expand file tree
/
Copy pathDFApp4.java
More file actions
122 lines (97 loc) · 4 KB
/
Copy pathDFApp4.java
File metadata and controls
122 lines (97 loc) · 4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
package com.example.df;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF3;
import org.apache.spark.sql.types.DataTypes;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;
public class DFApp4 {
static class WordDistanceUdf implements UDF3<String, String, String, Integer>{
private Pattern wordSep;
public WordDistanceUdf(){
wordSep = Pattern.compile("\\s+");
}
@Override
public Integer call(String text, String word1, String word2) throws Exception {
Integer result = null;
if(!(text == null || word1 == null || word2 == null)){
if(word1.equals(word2)){
result = 0;
}else { // 1. fast fast slow 2. fast fast fast
final String[] split = wordSep.split(text);
for (int i = 0; i < split.length; i++) {
if (split[i].equals(word1)) {
for (int j = 0; j < split.length; j++) {
if (split[j].equals(word2)) {
int d = Math.abs(i - j) - 1;
result = Math.min(result, d);
break;
}
}
}
}
}
}
return result;
}
}
public void start(String ...args){
SparkConf conf = new SparkConf();
conf.setAppName(getClass().getName());
conf.setMaster("local[4]"); // 4 executor threads
conf.setIfMissing("spark.master", "local[4]");
conf.setIfMissing("spark.default.parallelism", "16");
conf.set("spark.hadoop.validateOutputSpecs", "false");
conf.set("spark.eventLog.enabled", "true");
conf.set("spark.eventLog.dir", "/tmp/spark-events-logs");
final SparkSession sparkSession = SparkSession.builder().config(conf).getOrCreate();
/*
sparkSession.udf().register("word_distance", (String text, String word1, String word2) -> {
Integer result = null;
if(!(text == null || word1 == null || word2 == null)){
if(word1.equals(word2)){
result = 0;
}else {
final String[] split = text.split("\\s+");
for (int i = 0; i < split.length; i++) {
if (split[i].equals(word1)) {
for (int j = 0; j < split.length; j++) {
if (split[j].equals(word2)) {
result = Math.abs(i - j) - 1;
break;
}
}
}
}
}
}
return result;
}, DataTypes.IntegerType);
*/
sparkSession.udf().register("word_distance", new WordDistanceUdf(), DataTypes.IntegerType);
final String[] testStrings = new String[]{
"Fast text searching for regular expressions or automaton searching on tries",
null,
"hello world"
};
final Dataset<Row> dataset = sparkSession
.createDataset(Arrays.asList(testStrings), Encoders.STRING())
.toDF("value")
;
final List<Integer> distances = dataset
.selectExpr("word_distance(value, 'text', 'expressions')")
.as(Encoders.INT())
.collectAsList();
assert distances.get(0) == 3;
assert distances.get(1) == null;
assert distances.get(2) == null;
sparkSession.close();
}
public static void main(String ...args){
new DFApp4().start(args);
}
}