TPUAnalysis { public void run(String configPath, String outputPath, long timeMin, long timeMax, int limit)throws Exception { FlinkEnvManager fem = new FlinkEnvManager(configPath, "converterJob", TableIdentifier.RAW_TWITTER_DATA.get(),null); DataSet<Tuple2<Key,Value>> rawTwitterDataRows = fem.getDataFromAccumulo();; DataSet<Tuple2<String,Integer>> tweetCount = rawTwitterDataRows .filter(new TimeFilter(timeMin, timeMax)) .map(new UserTweetMap()) .groupBy(0) .sum(1); tweetCount =tweetCount.sortPartition(1, Order.DESCENDING) .setParallelism(1); if(limit >0){ tweetCount = tweetCount.first(limit).setParallelism(1); } tweetCount.writeAsCsv(outputPath).setParallelism(1); fem.getExecutionEnvironment().execute("TPUAnalysis"); }  void run(String configPath, String outputPath, long timeMin, long timeMax, int limit); static void main(String[] args);  }
@Test public void testConverterProcess() throws Exception { tmpSettingsDir.create(); File settings = tmpSettingsDir.newFile("settings"); File output = tmpSettingsDir.newFile("outputcsv"); output.delete(); Connector conn = amc.getConnector(); System.out.println("I am connected as: " + conn.whoami()); conn.tableOperations().create("RawTwitterData"); conn.tableOperations().create("TermIndex"); BatchWriter bw = conn.createBatchWriter("RawTwitterData", new BatchWriterConfig()); System.out.println("creating test table"); for(int i =5; i < 10;i++){ System.out.println("entry "+i); long ts1 = i; ByteBuffer bb = ByteBuffer.allocate(Long.BYTES + Integer.BYTES); bb.putLong(ts1); Mutation m1 = new Mutation(bb.array()); m1.put("CF", "CQ", "{\"text\":\"example tweet"+i+"\"," + "\"user\":{\"screen_name\":\"user1\", \"name\":\"u1name\"}} "); bw.addMutation(m1); } ByteBuffer bb = ByteBuffer.allocate(Long.BYTES + Integer.BYTES); bb.putLong(99); Mutation m1 = new Mutation(bb.array()); m1.put("CF", "CQ", "{\"text\":\"example tweet"+7+"\"," + "\"user\":{\"screen_name\":\"otherName\", \"name\":\"otherName\"}} "); bw.addMutation(m1); bw.close(); System.out.println("creating settings file"); FileOutputStream fos = new FileOutputStream(settings, false); BufferedWriter br = new BufferedWriter(new OutputStreamWriter(fos)); String parameters = "accumulo.instance=" + conn.getInstance().getInstanceName() + "\n"+ "accumulo.user=" + conn.whoami() +"\n"+ "accumulo.password=password\n"+ "accumulo.zookeeper=" + conn.getInstance().getZooKeepers(); System.out.println(parameters); br.write(parameters); br.flush(); br.close(); fos.flush(); fos.close(); TPUAnalysis tpua = new TPUAnalysis(); tpua.run(settings.getAbsolutePath(),output.getAbsolutePath(),0,100,0); System.out.println("test table:-----------------------------------------------------"); org.apache.accumulo.core.client.Scanner s = conn.createScanner("RawTwitterData", new Authorizations("standard")); for(Map.Entry<Key, Value> entry: s){ System.out.println(entry.getKey() + " | " +entry.getValue()); } s.close(); assertFalse(output.isDirectory()); List<String> resList =new ArrayList<>(); System.out.println("Output file:----------------------------------------------------"); Scanner in = new Scanner(new FileReader(output.getAbsolutePath())); while (in.hasNext()) { String str = in.next(); resList.add(str); System.out.println(str); } System.out.println("------------------------------------------------------------------"); assertEquals(2,resList.size()); assertEquals("u1name,5", resList.get(0)); assertEquals("otherName,1", resList.get(1)); }