@Test(expected=ServiceFacadeException.class) public void testRunWithoutISLookupFacade() throws Exception { parameters.remove(IMPORT_FACADE_FACTORY_CLASS); importer.run(portBindings, conf, parameters); }
@Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_IMPORT_CONTEXT_IDS_CSV), "unknown context identifier, required parameter '%s' is missing!", PARAM_IMPORT_CONTEXT_IDS_CSV); String contextIdsCSV = parameters.get(PARAM_IMPORT_CONTEXT_IDS_CSV); ISLookupFacade isLookupFacade = ServiceFacadeUtils.instantiate(parameters); try (DataFileWriter<Concept> conceptWriter = getWriter(FileSystem.get(conf), portBindings)) { SAXParserFactory parserFactory = SAXParserFactory.newInstance(); SAXParser saxParser = parserFactory.newSAXParser(); NamedCounters counters = new NamedCounters(new String[] { CONCEPT_COUNTER_NAME }); int count = 0; for (String contextXML : isLookupFacade.searchProfile(buildQuery(contextIdsCSV))) { count++; if (!StringUtils.isEmpty(contextXML)) { DataFileRecordReceiverWithCounter<Concept> conceptReciever = new DataFileRecordReceiverWithCounter<Concept>(conceptWriter); saxParser.parse(new InputSource(new StringReader(contextXML)), new ConceptXmlHandler(conceptReciever)); counters.increment(CONCEPT_COUNTER_NAME, conceptReciever.getReceivedCount()); } else { log.error("got empty context when looking for for context ids: " + contextIdsCSV); } } if (count==0) { log.warn("got 0 profiles when looking for context ids: " + contextIdsCSV); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } }
ISLookupServiceBasedConceptImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_IMPORT_CONTEXT_IDS_CSV), "unknown context identifier, required parameter '%s' is missing!", PARAM_IMPORT_CONTEXT_IDS_CSV); String contextIdsCSV = parameters.get(PARAM_IMPORT_CONTEXT_IDS_CSV); ISLookupFacade isLookupFacade = ServiceFacadeUtils.instantiate(parameters); try (DataFileWriter<Concept> conceptWriter = getWriter(FileSystem.get(conf), portBindings)) { SAXParserFactory parserFactory = SAXParserFactory.newInstance(); SAXParser saxParser = parserFactory.newSAXParser(); NamedCounters counters = new NamedCounters(new String[] { CONCEPT_COUNTER_NAME }); int count = 0; for (String contextXML : isLookupFacade.searchProfile(buildQuery(contextIdsCSV))) { count++; if (!StringUtils.isEmpty(contextXML)) { DataFileRecordReceiverWithCounter<Concept> conceptReciever = new DataFileRecordReceiverWithCounter<Concept>(conceptWriter); saxParser.parse(new InputSource(new StringReader(contextXML)), new ConceptXmlHandler(conceptReciever)); counters.increment(CONCEPT_COUNTER_NAME, conceptReciever.getReceivedCount()); } else { log.error("got empty context when looking for for context ids: " + contextIdsCSV); } } if (count==0) { log.warn("got 0 profiles when looking for context ids: " + contextIdsCSV); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } }
ISLookupServiceBasedConceptImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_IMPORT_CONTEXT_IDS_CSV), "unknown context identifier, required parameter '%s' is missing!", PARAM_IMPORT_CONTEXT_IDS_CSV); String contextIdsCSV = parameters.get(PARAM_IMPORT_CONTEXT_IDS_CSV); ISLookupFacade isLookupFacade = ServiceFacadeUtils.instantiate(parameters); try (DataFileWriter<Concept> conceptWriter = getWriter(FileSystem.get(conf), portBindings)) { SAXParserFactory parserFactory = SAXParserFactory.newInstance(); SAXParser saxParser = parserFactory.newSAXParser(); NamedCounters counters = new NamedCounters(new String[] { CONCEPT_COUNTER_NAME }); int count = 0; for (String contextXML : isLookupFacade.searchProfile(buildQuery(contextIdsCSV))) { count++; if (!StringUtils.isEmpty(contextXML)) { DataFileRecordReceiverWithCounter<Concept> conceptReciever = new DataFileRecordReceiverWithCounter<Concept>(conceptWriter); saxParser.parse(new InputSource(new StringReader(contextXML)), new ConceptXmlHandler(conceptReciever)); counters.increment(CONCEPT_COUNTER_NAME, conceptReciever.getReceivedCount()); } else { log.error("got empty context when looking for for context ids: " + contextIdsCSV); } } if (count==0) { log.warn("got 0 profiles when looking for context ids: " + contextIdsCSV); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } ISLookupServiceBasedConceptImporter(); }
ISLookupServiceBasedConceptImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_IMPORT_CONTEXT_IDS_CSV), "unknown context identifier, required parameter '%s' is missing!", PARAM_IMPORT_CONTEXT_IDS_CSV); String contextIdsCSV = parameters.get(PARAM_IMPORT_CONTEXT_IDS_CSV); ISLookupFacade isLookupFacade = ServiceFacadeUtils.instantiate(parameters); try (DataFileWriter<Concept> conceptWriter = getWriter(FileSystem.get(conf), portBindings)) { SAXParserFactory parserFactory = SAXParserFactory.newInstance(); SAXParser saxParser = parserFactory.newSAXParser(); NamedCounters counters = new NamedCounters(new String[] { CONCEPT_COUNTER_NAME }); int count = 0; for (String contextXML : isLookupFacade.searchProfile(buildQuery(contextIdsCSV))) { count++; if (!StringUtils.isEmpty(contextXML)) { DataFileRecordReceiverWithCounter<Concept> conceptReciever = new DataFileRecordReceiverWithCounter<Concept>(conceptWriter); saxParser.parse(new InputSource(new StringReader(contextXML)), new ConceptXmlHandler(conceptReciever)); counters.increment(CONCEPT_COUNTER_NAME, conceptReciever.getReceivedCount()); } else { log.error("got empty context when looking for for context ids: " + contextIdsCSV); } } if (count==0) { log.warn("got 0 profiles when looking for context ids: " + contextIdsCSV); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } ISLookupServiceBasedConceptImporter(); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf,
			Map<String, String> parameters); }
ISLookupServiceBasedConceptImporter implements Process { @Override public void run(PortBindings portBindings, Configuration conf, Map<String, String> parameters) throws Exception { Preconditions.checkArgument(parameters.containsKey(PARAM_IMPORT_CONTEXT_IDS_CSV), "unknown context identifier, required parameter '%s' is missing!", PARAM_IMPORT_CONTEXT_IDS_CSV); String contextIdsCSV = parameters.get(PARAM_IMPORT_CONTEXT_IDS_CSV); ISLookupFacade isLookupFacade = ServiceFacadeUtils.instantiate(parameters); try (DataFileWriter<Concept> conceptWriter = getWriter(FileSystem.get(conf), portBindings)) { SAXParserFactory parserFactory = SAXParserFactory.newInstance(); SAXParser saxParser = parserFactory.newSAXParser(); NamedCounters counters = new NamedCounters(new String[] { CONCEPT_COUNTER_NAME }); int count = 0; for (String contextXML : isLookupFacade.searchProfile(buildQuery(contextIdsCSV))) { count++; if (!StringUtils.isEmpty(contextXML)) { DataFileRecordReceiverWithCounter<Concept> conceptReciever = new DataFileRecordReceiverWithCounter<Concept>(conceptWriter); saxParser.parse(new InputSource(new StringReader(contextXML)), new ConceptXmlHandler(conceptReciever)); counters.increment(CONCEPT_COUNTER_NAME, conceptReciever.getReceivedCount()); } else { log.error("got empty context when looking for for context ids: " + contextIdsCSV); } } if (count==0) { log.warn("got 0 profiles when looking for context ids: " + contextIdsCSV); } countersWriter.writeCounters(counters, System.getProperty(OOZIE_ACTION_OUTPUT_FILENAME)); } } ISLookupServiceBasedConceptImporter(); @Override Map<String, PortType> getInputPorts(); @Override Map<String, PortType> getOutputPorts(); @Override void run(PortBindings portBindings, Configuration conf,
			Map<String, String> parameters); static final String PARAM_IMPORT_CONTEXT_IDS_CSV; }