@Test public void testRun() throws Exception { importer.run(portBindings, conf, parameters); verify(projectWriter, times(2)).append(projectCaptor.capture()); List<Project> projects = projectCaptor.getAllValues(); assertEquals(2, projects.size()); assertNotNull(projects.get(0)); assertNotNull(projects.get(1)); verifyReport(2, PROJECT_COUNTER_NAME); }
@Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { FileSystem fs = FileSystem.get(conf); try (DataFileWriter<Project> projectWriter = getWriter(fs, portBindings)) { NamedCounters counters = new NamedCounters(new String[] { PROJECT_COUNTER_NAME }); StreamingFacade streamingFacade = ServiceFacadeUtils.instantiate(parameters); try (Scanner scanner = new Scanner(new BufferedInputStream(streamingFacade.getStream()), "utf8")) { ProjectDetailConverter converter = new ProjectDetailConverter(); int currentCount = 0; long startTime = System.currentTimeMillis(); while (scanner.hasNext()) { String line = scanner.nextLine(); if (StringUtils.isNotBlank(line)) { ProjectDetail project = ProjectDetail.fromJson(line); projectWriter.append(converter.convert(project)); counters.increment(PROJECT_COUNTER_NAME); currentCount++; if (currentCount%progressLogInterval==0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis()-startTime)/1000) + " secs"); startTime = System.currentTimeMillis(); } } } } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } }
StreamingProjectImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { FileSystem fs = FileSystem.get(conf); try (DataFileWriter<Project> projectWriter = getWriter(fs, portBindings)) { NamedCounters counters = new NamedCounters(new String[] { PROJECT_COUNTER_NAME }); StreamingFacade streamingFacade = ServiceFacadeUtils.instantiate(parameters); try (Scanner scanner = new Scanner(new BufferedInputStream(streamingFacade.getStream()), "utf8")) { ProjectDetailConverter converter = new ProjectDetailConverter(); int currentCount = 0; long startTime = System.currentTimeMillis(); while (scanner.hasNext()) { String line = scanner.nextLine(); if (StringUtils.isNotBlank(line)) { ProjectDetail project = ProjectDetail.fromJson(line); projectWriter.append(converter.convert(project)); counters.increment(PROJECT_COUNTER_NAME); currentCount++; if (currentCount%progressLogInterval==0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis()-startTime)/1000) + " secs"); startTime = System.currentTimeMillis(); } } } } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } }
StreamingProjectImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { FileSystem fs = FileSystem.get(conf); try (DataFileWriter<Project> projectWriter = getWriter(fs, portBindings)) { NamedCounters counters = new NamedCounters(new String[] { PROJECT_COUNTER_NAME }); StreamingFacade streamingFacade = ServiceFacadeUtils.instantiate(parameters); try (Scanner scanner = new Scanner(new BufferedInputStream(streamingFacade.getStream()), "utf8")) { ProjectDetailConverter converter = new ProjectDetailConverter(); int currentCount = 0; long startTime = System.currentTimeMillis(); while (scanner.hasNext()) { String line = scanner.nextLine(); if (StringUtils.isNotBlank(line)) { ProjectDetail project = ProjectDetail.fromJson(line); projectWriter.append(converter.convert(project)); counters.increment(PROJECT_COUNTER_NAME); currentCount++; if (currentCount%progressLogInterval==0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis()-startTime)/1000) + " secs"); startTime = System.currentTimeMillis(); } } } } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } StreamingProjectImporter(); }
StreamingProjectImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { FileSystem fs = FileSystem.get(conf); try (DataFileWriter<Project> projectWriter = getWriter(fs, portBindings)) { NamedCounters counters = new NamedCounters(new String[] { PROJECT_COUNTER_NAME }); StreamingFacade streamingFacade = ServiceFacadeUtils.instantiate(parameters); try (Scanner scanner = new Scanner(new BufferedInputStream(streamingFacade.getStream()), "utf8")) { ProjectDetailConverter converter = new ProjectDetailConverter(); int currentCount = 0; long startTime = System.currentTimeMillis(); while (scanner.hasNext()) { String line = scanner.nextLine(); if (StringUtils.isNotBlank(line)) { ProjectDetail project = ProjectDetail.fromJson(line); projectWriter.append(converter.convert(project)); counters.increment(PROJECT_COUNTER_NAME); currentCount++; if (currentCount%progressLogInterval==0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis()-startTime)/1000) + " secs"); startTime = System.currentTimeMillis(); } } } } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } StreamingProjectImporter(); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters); }
StreamingProjectImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { FileSystem fs = FileSystem.get(conf); try (DataFileWriter<Project> projectWriter = getWriter(fs, portBindings)) { NamedCounters counters = new NamedCounters(new String[] { PROJECT_COUNTER_NAME }); StreamingFacade streamingFacade = ServiceFacadeUtils.instantiate(parameters); try (Scanner scanner = new Scanner(new BufferedInputStream(streamingFacade.getStream()), "utf8")) { ProjectDetailConverter converter = new ProjectDetailConverter(); int currentCount = 0; long startTime = System.currentTimeMillis(); while (scanner.hasNext()) { String line = scanner.nextLine(); if (StringUtils.isNotBlank(line)) { ProjectDetail project = ProjectDetail.fromJson(line); projectWriter.append(converter.convert(project)); counters.increment(PROJECT_COUNTER_NAME); currentCount++; if (currentCount%progressLogInterval==0) { log.info("current progress: " + currentCount + ", last package of " + progressLogInterval + " processed in " + ((System.currentTimeMillis()-startTime)/1000) + " secs"); startTime = System.currentTimeMillis(); } } } } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } StreamingProjectImporter(); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters);  }