0
Je veux implémenter le modèle de Markov dans Flink. Premièrement, j'ai lu des données de Kafka. Comment je peux implémenter le modèle de trigram Markov avec le flink?Mise en œuvre du modèle Markov Flink
Je veux implémenter le modèle de Markov dans Flink. Premièrement, j'ai lu des données de Kafka. Comment je peux implémenter le modèle de trigram Markov avec le flink?Mise en œuvre du modèle Markov Flink
J'implante enfin le modèle markov. Ce code calcule la matrice de transition uniquement.
private static class MarkovModel implements AllWindowFunction<Tuple2<String,String>, Tuple3<Long, Long, HashMap<String,Integer>>, TimeWindow>{
@Override
public void apply(TimeWindow window, Iterable<Tuple2<String, String>> requests, Collector<Tuple3<Long, Long, HashMap<String, Integer>>> out) throws Exception {
HashMap<String,Integer> map = new HashMap<>();
String first = "";
String second = "";
String third = "";
for (Tuple2<String, String> request : requests) {
if(first == ""){
third = second;
second = first;
first = request.f1;
}else if(second == ""){
third = second;
second = request.f1;
}else if(third == ""){
third = request.f1;
}else{
third = second;
second = first;
first = request.f1;
}
if(third != ""){
int count = map.getOrDefault(first + second + third,0);
map.put(first + second + third,count + 1);
}
}
System.out.println(map);
System.out.println(map.values().stream().mapToDouble(x->x).sum());
out.collect(new Tuple3(window.getStart(), window.getEnd(), map));
}
}