@Test public void testTableLineage() { final String processorName = "PutHiveStreaming"; final String transitUri = "thrift: final ProvenanceEventRecord record = Mockito.mock(ProvenanceEventRecord.class); when(record.getComponentType()).thenReturn(processorName); when(record.getTransitUri()).thenReturn(transitUri); when(record.getEventType()).thenReturn(ProvenanceEventType.SEND); when(record.getAttribute(ATTR_OUTPUT_TABLES)).thenReturn("databaseA.tableA"); final ClusterResolvers clusterResolvers = Mockito.mock(ClusterResolvers.class); when(clusterResolvers.fromHostNames(matches(".+\\.example\\.com"))).thenReturn("cluster1"); final AnalysisContext context = Mockito.mock(AnalysisContext.class); when(context.getClusterResolver()).thenReturn(clusterResolvers); final NiFiProvenanceEventAnalyzer analyzer = NiFiProvenanceEventAnalyzerFactory.getAnalyzer(processorName, transitUri, record.getEventType()); assertNotNull(analyzer); final DataSetRefs refs = analyzer.analyze(context, record); assertEquals(0, refs.getInputs().size()); assertEquals(1, refs.getOutputs().size()); Referenceable ref = refs.getOutputs().iterator().next(); assertEquals("hive_table", ref.getTypeName()); assertEquals("tableA", ref.get(ATTR_NAME)); assertEquals("databaseA.tableA@cluster1", ref.get(ATTR_QUALIFIED_NAME)); }
@Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final Set<Tuple<String, String>> outputTables = parseTableNames(null, event.getAttribute(ATTR_OUTPUT_TABLES)); if (outputTables.isEmpty()) { return null; } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); outputTables.forEach(tableName -> { final Referenceable ref = createTableRef(clusterName, tableName); refs.addOutput(ref); }); return refs; }
PutHiveStreaming extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final Set<Tuple<String, String>> outputTables = parseTableNames(null, event.getAttribute(ATTR_OUTPUT_TABLES)); if (outputTables.isEmpty()) { return null; } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); outputTables.forEach(tableName -> { final Referenceable ref = createTableRef(clusterName, tableName); refs.addOutput(ref); }); return refs; } }
PutHiveStreaming extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final Set<Tuple<String, String>> outputTables = parseTableNames(null, event.getAttribute(ATTR_OUTPUT_TABLES)); if (outputTables.isEmpty()) { return null; } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); outputTables.forEach(tableName -> { final Referenceable ref = createTableRef(clusterName, tableName); refs.addOutput(ref); }); return refs; }  }
PutHiveStreaming extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final Set<Tuple<String, String>> outputTables = parseTableNames(null, event.getAttribute(ATTR_OUTPUT_TABLES)); if (outputTables.isEmpty()) { return null; } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); outputTables.forEach(tableName -> { final Referenceable ref = createTableRef(clusterName, tableName); refs.addOutput(ref); }); return refs; }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetComponentTypePattern(); }
PutHiveStreaming extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final Set<Tuple<String, String>> outputTables = parseTableNames(null, event.getAttribute(ATTR_OUTPUT_TABLES)); if (outputTables.isEmpty()) { return null; } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); outputTables.forEach(tableName -> { final Referenceable ref = createTableRef(clusterName, tableName); refs.addOutput(ref); }); return refs; }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetComponentTypePattern();  }