ConverterFlatMap implements FlatMapFunction<Tuple2<Key, Value>, Tuple2<Text,Mutation>>, Serializable { @Override public void flatMap(Tuple2<Key, Value> in, Collector<Tuple2<Text, Mutation>> out) { JSONObject obj = null; String userName = ""; String userScreenName = ""; String text = ""; try { obj = new JSONObject(in.f1.toString()); text = obj.getString("text"); userScreenName = obj.getJSONObject("user").getString("screen_name"); userName = obj.getJSONObject("user").getString("name"); } catch (JSONException e) { e.printStackTrace(); } Map<String, Integer> tokenCount= new HashMap<>(); for(String token: tokenizer.tokenizeString(text)){ if(tokenCount.containsKey(token)){ tokenCount.put(token,tokenCount.get(token)+1); }else{ tokenCount.put(token,1); } } for(Map.Entry<String, Integer> kv: tokenCount.entrySet()){ Mutation m = new Mutation(kv.getKey()); if(kv.getValue() > 1){ m.put("text", in.f0.getRow().toString(), kv.getValue().toString()); }else{ m.put("text", in.f0.getRow().toString(), new Value()); } out.collect(new Tuple2<>(new Text(outputTableName), m)); } Mutation m = new Mutation(userName.toLowerCase()); m.put("user", in.f0.getRow().toString(), new Value()); out.collect(new Tuple2<>(new Text(outputTableName), m)); m = new Mutation(userScreenName.toLowerCase()); m.put("user", in.f0.getRow().toString(), new Value()); out.collect(new Tuple2<>(new Text(outputTableName), m)); } ConverterFlatMap(Tokenizer tokenizer, String outputTableName); @Override void flatMap(Tuple2<Key, Value> in, Collector<Tuple2<Text, Mutation>> out);  }
@Test public void testFlatMap() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<Tuple2<Key,Value>> input = env.fromElements( new Tuple2<>(new Key("row1","t",""), new Value(("{\"text\":\"example tweet\"," + "\"user\":{\"screen_name\":\"user1\", \"name\":\"u1name\"}} ").getBytes()))); DataSet<Tuple2<Text,Mutation>> output = input.flatMap( new ConverterFlatMap(new Tokenizer(),"table")); output.print(); assertEquals(output.count(), 4); }