private void configureComponentsBeforeBuilding ( final AnalysisJobBuilder jobBuilder , final int partitionNumber ) { // update datastores and resource properties to point to node-specific // targets if possible. This way parallel writing to files on HDFS does // not cause any inconsistencies because each node is writing to a // separate file. for ( final ComponentBuilder cb : jobBuilder . getComponentBuilders ( ) ) { // find any datastore properties that point to HDFS files final Set < ConfiguredPropertyDescriptor > targetDatastoreProperties = cb . getDescriptor ( ) . getConfiguredPropertiesByType ( UpdateableDatastore . class , false ) ; for ( final ConfiguredPropertyDescriptor targetDatastoreProperty : targetDatastoreProperties ) { final Object datastoreObject = cb . getConfiguredProperty ( targetDatastoreProperty ) ; if ( datastoreObject instanceof ResourceDatastore ) { final ResourceDatastore resourceDatastore = ( ResourceDatastore ) datastoreObject ; final Resource resource = resourceDatastore . getResource ( ) ; final Resource replacementResource = createReplacementResource ( resource , partitionNumber ) ; if ( replacementResource != null ) { final ResourceDatastore replacementDatastore = createReplacementDatastore ( cb , resourceDatastore , replacementResource ) ; if ( replacementDatastore != null ) { cb . setConfiguredProperty ( targetDatastoreProperty , replacementDatastore ) ; } } } } final Set < ConfiguredPropertyDescriptor > resourceProperties = cb . getDescriptor ( ) . getConfiguredPropertiesByType ( Resource . class , false ) ; for ( final ConfiguredPropertyDescriptor resourceProperty : resourceProperties ) { final Resource resource = ( Resource ) cb . getConfiguredProperty ( resourceProperty ) ; final Resource replacementResource = createReplacementResource ( resource , partitionNumber ) ; if ( replacementResource != null ) { cb . setConfiguredProperty ( resourceProperty , replacementResource ) ; } } // special handlings of specific component types are handled here if ( cb . getComponentInstance ( ) instanceof CreateCsvFileAnalyzer ) { if ( partitionNumber > 0 ) { // ensure header is only created once cb . setConfiguredProperty ( CreateCsvFileAnalyzer . PROPERTY_INCLUDE_HEADER , false ) ; } } } // recursively apply this function also on output data stream jobs final List < AnalysisJobBuilder > children = jobBuilder . getConsumedOutputDataStreamsJobBuilders ( ) ; for ( final AnalysisJobBuilder childJobBuilder : children ) { configureComponentsBeforeBuilding ( childJobBuilder , partitionNumber ) ; } }
Applies any partition - specific configuration to the job builder before building it .