@Test public void testGetFlowTemplate() throws Exception { URI flowTemplateCatalogUri = this.getClass().getClassLoader().getResource("template_catalog").toURI(); Properties properties = new Properties(); properties.put(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY, flowTemplateCatalogUri.toString()); Config config = ConfigFactory.parseProperties(properties); Config templateCatalogCfg = config .withValue(ConfigurationKeys.JOB_CONFIG_FILE_GENERAL_PATH_KEY, config.getValue(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY)); FSFlowTemplateCatalog catalog = new FSFlowTemplateCatalog(templateCatalogCfg); FlowTemplate flowTemplate = catalog.getFlowTemplate(new URI(TEST_TEMPLATE_DIR_URI)); List<JobTemplate> jobTemplates = flowTemplate.getJobTemplates(); Assert.assertEquals(jobTemplates.size(), 4); for (int i = 0; i < 4; i++) { String uri = new Path(jobTemplates.get(i).getUri()).getName().split("\\.")[0]; String templateId = uri.substring(uri.length() - 1); for (int j = 0; j < 2; j++) { Config jobTemplateConfig = jobTemplates.get(i).getRawTemplateConfig(); String suffix = templateId + Integer.toString(j + 1); Assert.assertEquals(jobTemplateConfig.getString("key" + suffix), "val" + suffix); } } Config flowConfig = ConfigFactory.empty().withValue("team.name", ConfigValueFactory.fromAnyRef("test-team")) .withValue("dataset.name", ConfigValueFactory.fromAnyRef("test-dataset")); List<Pair<DatasetDescriptor, DatasetDescriptor>> inputOutputDescriptors = flowTemplate.getDatasetDescriptors(flowConfig, true); Assert.assertTrue(inputOutputDescriptors.size() == 2); List<String> dirs = Lists.newArrayList("inbound", "outbound"); for (int i = 0; i < 2; i++) { for (int j = 0; j < 2; j++) { FSDatasetDescriptor datasetDescriptor; if (j == 0) { datasetDescriptor = (FSDatasetDescriptor) inputOutputDescriptors.get(i).getLeft(); } else { datasetDescriptor = (FSDatasetDescriptor) inputOutputDescriptors.get(i).getRight(); } Assert.assertEquals(datasetDescriptor.getPlatform(), "hdfs"); Assert.assertEquals(datasetDescriptor.getFormatConfig().getFormat(), "avro"); Assert.assertEquals(datasetDescriptor.getPath(), "/data/" + dirs.get(i) + "/test-team/test-dataset"); } } Config flowTemplateConfig = flowTemplate.getRawTemplateConfig(); Assert.assertEquals(flowTemplateConfig.getString(DatasetDescriptorConfigKeys.FLOW_EDGE_INPUT_DATASET_DESCRIPTOR_PREFIX + ".0." + DatasetDescriptorConfigKeys.CLASS_KEY), FSDatasetDescriptor.class.getCanonicalName()); Assert.assertEquals(flowTemplateConfig.getString(DatasetDescriptorConfigKeys.FLOW_EDGE_OUTPUT_DATASET_DESCRIPTOR_PREFIX + ".0." + DatasetDescriptorConfigKeys.CLASS_KEY), FSDatasetDescriptor.class.getCanonicalName()); }
public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { if (!validateTemplateURI(flowTemplateDirURI)) { throw new JobTemplate.TemplateException(String.format("The FlowTemplate %s is not valid", flowTemplateDirURI)); } String templateCatalogDir = this.sysConfig.getString(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY); Path templateDirPath = PathUtils.mergePaths(new Path(templateCatalogDir), new Path(flowTemplateDirURI.getPath())); Path templateFullPath = PathUtils.mergePaths(templateDirPath, new Path(FLOW_CONF_FILE_NAME)); FileSystem fs = FileSystem.get(templateFullPath.toUri(), new Configuration()); try (InputStream is = fs.open(templateFullPath)) { return new HOCONInputStreamFlowTemplate(is, flowTemplateDirURI, this); } }
FSFlowTemplateCatalog extends FSJobCatalog implements FlowCatalogWithTemplates { public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { if (!validateTemplateURI(flowTemplateDirURI)) { throw new JobTemplate.TemplateException(String.format("The FlowTemplate %s is not valid", flowTemplateDirURI)); } String templateCatalogDir = this.sysConfig.getString(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY); Path templateDirPath = PathUtils.mergePaths(new Path(templateCatalogDir), new Path(flowTemplateDirURI.getPath())); Path templateFullPath = PathUtils.mergePaths(templateDirPath, new Path(FLOW_CONF_FILE_NAME)); FileSystem fs = FileSystem.get(templateFullPath.toUri(), new Configuration()); try (InputStream is = fs.open(templateFullPath)) { return new HOCONInputStreamFlowTemplate(is, flowTemplateDirURI, this); } } }
FSFlowTemplateCatalog extends FSJobCatalog implements FlowCatalogWithTemplates { public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { if (!validateTemplateURI(flowTemplateDirURI)) { throw new JobTemplate.TemplateException(String.format("The FlowTemplate %s is not valid", flowTemplateDirURI)); } String templateCatalogDir = this.sysConfig.getString(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY); Path templateDirPath = PathUtils.mergePaths(new Path(templateCatalogDir), new Path(flowTemplateDirURI.getPath())); Path templateFullPath = PathUtils.mergePaths(templateDirPath, new Path(FLOW_CONF_FILE_NAME)); FileSystem fs = FileSystem.get(templateFullPath.toUri(), new Configuration()); try (InputStream is = fs.open(templateFullPath)) { return new HOCONInputStreamFlowTemplate(is, flowTemplateDirURI, this); } } FSFlowTemplateCatalog(Config sysConfig); }
FSFlowTemplateCatalog extends FSJobCatalog implements FlowCatalogWithTemplates { public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { if (!validateTemplateURI(flowTemplateDirURI)) { throw new JobTemplate.TemplateException(String.format("The FlowTemplate %s is not valid", flowTemplateDirURI)); } String templateCatalogDir = this.sysConfig.getString(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY); Path templateDirPath = PathUtils.mergePaths(new Path(templateCatalogDir), new Path(flowTemplateDirURI.getPath())); Path templateFullPath = PathUtils.mergePaths(templateDirPath, new Path(FLOW_CONF_FILE_NAME)); FileSystem fs = FileSystem.get(templateFullPath.toUri(), new Configuration()); try (InputStream is = fs.open(templateFullPath)) { return new HOCONInputStreamFlowTemplate(is, flowTemplateDirURI, this); } } FSFlowTemplateCatalog(Config sysConfig); FlowTemplate getFlowTemplate(URI flowTemplateDirURI); List<JobTemplate> getJobTemplatesForFlow(URI flowTemplateDirURI); boolean getAndSetShouldRefreshFlowGraph(boolean value); }
FSFlowTemplateCatalog extends FSJobCatalog implements FlowCatalogWithTemplates { public FlowTemplate getFlowTemplate(URI flowTemplateDirURI) throws SpecNotFoundException, JobTemplate.TemplateException, IOException, URISyntaxException { if (!validateTemplateURI(flowTemplateDirURI)) { throw new JobTemplate.TemplateException(String.format("The FlowTemplate %s is not valid", flowTemplateDirURI)); } String templateCatalogDir = this.sysConfig.getString(ServiceConfigKeys.TEMPLATE_CATALOGS_FULLY_QUALIFIED_PATH_KEY); Path templateDirPath = PathUtils.mergePaths(new Path(templateCatalogDir), new Path(flowTemplateDirURI.getPath())); Path templateFullPath = PathUtils.mergePaths(templateDirPath, new Path(FLOW_CONF_FILE_NAME)); FileSystem fs = FileSystem.get(templateFullPath.toUri(), new Configuration()); try (InputStream is = fs.open(templateFullPath)) { return new HOCONInputStreamFlowTemplate(is, flowTemplateDirURI, this); } } FSFlowTemplateCatalog(Config sysConfig); FlowTemplate getFlowTemplate(URI flowTemplateDirURI); List<JobTemplate> getJobTemplatesForFlow(URI flowTemplateDirURI); boolean getAndSetShouldRefreshFlowGraph(boolean value); static final String JOBS_DIR_NAME; static final String FLOW_CONF_FILE_NAME; static final List<String> JOB_FILE_EXTENSIONS; }