ConverterProcess { public void run(String configPath) throws Exception { FlinkEnvManager fem = new FlinkEnvManager(configPath, "converterJob", TableIdentifier.RAW_TWITTER_DATA.get(), TableIdentifier.TERM_INDEX.get()); DataSet<Tuple2<Key,Value>> rawTwitterDataRows = fem.getDataFromAccumulo(); DataSet<Tuple2<Text, Mutation>> termIndexMutations = rawTwitterDataRows .flatMap(new ConverterFlatMap(new Tokenizer(), TableIdentifier.TERM_INDEX.get())); termIndexMutations.output(fem.getHadoopOF()); fem.getExecutionEnvironment().execute("ConverterProcess"); }  void run(String configPath); static void main(String[] args);  }
@Test public void testConverterProcess() throws Exception { tmpSettingsDir.create(); File settings = tmpSettingsDir.newFile("settings"); Connector conn = amc.getConnector(); System.out.println("I am connected as: " + conn.whoami()); conn.tableOperations().create("RawTwitterData"); conn.tableOperations().create("TermIndex"); Mutation m1 = new Mutation("row1"); m1.put("CF", "CQ", "{\n" + " \"text\": \"RT @PostGradProblem: In preparation for for for the NFL lockout, I will be spending twice as much time analyzing my fantasy baseball team during ...\", \n\"user\": {\n" + " \"notifications\": null, \n" + " \"profile_use_background_image\": true, \n" + " \"statuses_count\": 351, \n" + " \"profile_background_color\": \"C0DEED\", \n" + " \"followers_count\": 48, \n" + " \"profile_image_url\": \"http: " \"listed_count\": 0, \n" + " \"profile_background_image_url\": \"http: " \"description\": \"watcha doin in my waters?\", \n" + " \"screen_name\": \"OldGREG85\", \n" + " \"default_profile\": true, \n" + " \"verified\": false, \n" + " \"time_zone\": \"Hawaii\", \n" + " \"profile_text_color\": \"333333\", \n" + " \"is_translator\": false, \n" + " \"profile_sidebar_fill_color\": \"DDEEF6\", \n" + " \"location\": \"Texas\", \n" + " \"id_str\": \"80177619\", \n" + " \"default_profile_image\": false, \n" + " \"profile_background_tile\": false, \n" + " \"lang\": \"en\", \n" + " \"friends_count\": 81, \n" + " \"protected\": false, \n" + " \"favourites_count\": 0, \n" + " \"created_at\": \"Tue Oct 06 01:13:17 +0000 2009\", \n" + " \"profile_link_color\": \"0084B4\", \n" + " \"name\": \"GG\", \n" + " \"show_all_inline_media\": false, \n" + " \"follow_request_sent\": null, \n" + " \"geo_enabled\": false, \n" + " \"profile_sidebar_border_color\": \"C0DEED\", \n" + " \"url\": null, \n" + " \"id\": 80177619, \n" + " \"contributors_enabled\": false, \n" + " \"following\": null, \n" + " \"utc_offset\": -36000\n" + " }}"); BatchWriter bw = conn.createBatchWriter("RawTwitterData", new BatchWriterConfig()); bw.addMutation(m1); bw.close(); 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(); ConverterProcess d = new ConverterProcess(); System.out.println("settings file path: "+settings.getAbsolutePath()); d.run(settings.getAbsolutePath()); System.out.println("RawTwitterData: -----------------------------------------------------"); Scanner s = conn.createScanner("RawTwitterData", new Authorizations("standard")); for(Map.Entry<Key, Value> entry: s){ System.out.println(entry.getKey() + " | " +entry.getValue()); } s.close(); System.out.println("TermIndex: -----------------------------------------------------"); s = conn.createScanner("TermIndex", new Authorizations("standard")); int i = 0; for(Map.Entry<Key, Value> entry: s){ System.out.println(entry.getKey() + " | " + entry.getValue()); i++; if(entry.getKey().getRow().toString().equals("for")){ assertEquals(entry.getValue().toString(),"3"); } } s.close(); System.out.println("TermIndex has "+i+" entrys"); d = new ConverterProcess(); System.out.println("settings file path: "+settings.getAbsolutePath()); d.run(settings.getAbsolutePath()); System.out.println("RawTwitterData: -----------------------------------------------------"); s = conn.createScanner("RawTwitterData", new Authorizations("standard")); for(Map.Entry<Key, Value> entry: s){ System.out.println(entry.getKey() + " | " +entry.getValue()); } s.close(); System.out.println("TermIndex: -----------------------------------------------------"); s = conn.createScanner("TermIndex", new Authorizations("standard")); i = 0; for(Map.Entry<Key, Value> entry: s){ System.out.println(entry.getKey() + " | " + entry.getValue()); i++; if(entry.getKey().getRow().toString().equals("for")){ assertEquals(entry.getValue().toString(),"3"); } } s.close(); }