UserTweetMap implements MapFunction<Tuple2<Key, Value>, Tuple2<String,Integer>> { @Override public Tuple2<String, Integer> map(Tuple2<Key, Value> in) throws Exception { JSONObject obj = null; String userName = ""; try { obj = new JSONObject(in.f1.toString()); userName = obj.getJSONObject("user").getString("name"); } catch (JSONException e) { e.printStackTrace(); } return new Tuple2<>(userName,1); }  @Override Tuple2<String, Integer> map(Tuple2<Key, Value> in);  }
@Test public void testUserTweetMap() throws Exception { final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<Tuple2<Key,Value>> data = env.fromElements( new Tuple2<>(new Key(), new Value((new Text("{\"text\":\"example tweet\"," + "\"user\":{\"screen_name\":\"user1\", \"name\":\"u1name\"}} ")).getBytes())), new Tuple2<>(new Key(), new Value((new Text("{\"text\":\"example tweet\"," + "\"user\":{\"screen_name\":\"user1\", \"name\":\"u1name\"}} ")).getBytes())), new Tuple2<>(new Key(), new Value((new Text("{\"text\":\"example tweet\"," + "\"user\":{\"screen_name\":\"user2\", \"name\":\"u1name2\"}} ")).getBytes()))); DataSet<Tuple2<String,Integer>> out= data.map(new UserTweetMap()); out.print(); assertEquals(3,out.count()); }