Package org.apache.nifi.processors.standard

Class MergeContent

java.lang.Object
org.apache.nifi.components.AbstractConfigurableComponent
org.apache.nifi.processor.AbstractSessionFactoryProcessor
org.apache.nifi.processor.util.bin.BinFiles
org.apache.nifi.processors.standard.MergeContent
All Implemented Interfaces:
org.apache.nifi.components.ConfigurableComponent, org.apache.nifi.processor.Processor
@SideEffectFree @TriggerWhenEmpty @InputRequirement(INPUT_REQUIRED) @Tags({"merge","content","correlation","tar","zip","stream","concatenation","archive","flowfile-stream","flowfile-stream-v3"}) @CapabilityDescription("Merges a Group of FlowFiles together based on a user-defined strategy and packages them into a single FlowFile. It is recommended that the Processor be configured with only a single incoming connection, as Group of FlowFiles will not be created from FlowFiles in different connections. This processor updates the mime.type attribute as appropriate. NOTE: this processor should NOT be configured with Cron Driven for the Scheduling Strategy.") @ReadsAttribute(attribute="fragment.identifier",description="Applicable only if the <Merge Strategy> property is set to Defragment. All FlowFiles with the same value for this attribute will be bundled together.") @ReadsAttribute(attribute="fragment.index",description="Applicable only if the <Merge Strategy> property is set to Defragment. This attribute indicates the order in which the fragments should be assembled. This attribute must be present on all FlowFiles when using the Defragment Merge Strategy and must be a unique (i.e., unique across all FlowFiles that have the same value for the \"fragment.identifier\" attribute) integer between 0 and the value of the fragment.count attribute. If two or more FlowFiles have the same value for the \"fragment.identifier\" attribute and the same value for the \"fragment.index\" attribute, the first FlowFile processed will be accepted and subsequent FlowFiles will not be accepted into the Bin.") @ReadsAttribute(attribute="fragment.count",description="Applicable only if the <Merge Strategy> property is set to Defragment. This attribute indicates how many FlowFiles should be expected in the given bundle. At least one FlowFile must have this attribute in the bundle. If multiple FlowFiles contain the \"fragment.count\" attribute in a given bundle, all must have the same value.") @ReadsAttribute(attribute="segment.original.filename",description="Applicable only if the <Merge Strategy> property is set to Defragment. This attribute must be present on all FlowFiles with the same value for the fragment.identifier attribute. All FlowFiles in the same bundle must have the same value for this attribute. The value of this attribute will be used for the filename of the completed merged FlowFile.") @ReadsAttribute(attribute="tar.permissions",description="Applicable only if the <Merge Format> property is set to TAR. The value of this attribute must be 3 characters; each character must be in the range 0 to 7 (inclusive) and indicates the file permissions that should be used for the FlowFile\'s TAR entry. If this attribute is missing or has an invalid value, the default value of 644 will be used") @WritesAttribute(attribute="filename",description="When more than 1 file is merged, the filename comes from the segment.original.filename attribute. If that attribute does not exist in the source FlowFiles, then the filename is set to the number of nanoseconds matching system time. Then a filename extension may be applied:if Merge Format is TAR, then the filename will be appended with .tar, if Merge Format is ZIP, then the filename will be appended with .zip, if Merge Format is FlowFileStream, then the filename will be appended with .pkg") @WritesAttribute(attribute="merge.count",description="The number of FlowFiles that were merged into this bundle") @WritesAttribute(attribute="merge.bin.age",description="The age of the bin, in milliseconds, when it was merged and output. Effectively this is the greatest amount of time that any FlowFile in this bundle remained waiting in this processor before it was output") @WritesAttribute(attribute="merge.uuid",description="UUID of the merged flow file that will be added to the original flow files attributes.") @WritesAttribute(attribute="merge.reason",description="This processor allows for several thresholds to be configured for merging FlowFiles. This attribute indicates which of the Thresholds resulted in the FlowFiles being merged. For an explanation of each of the possible values and their meanings, see the Processor\'s Usage / documentation and see the \'Additional Details\' page.") @SeeAlso({SegmentContent.class,MergeRecord.class}) @SystemResourceConsideration(resource=MEMORY, description="While content is not stored in memory, the FlowFiles\' attributes are. The configuration of MergeContent (maximum bin size, maximum group size, maximum bin age, max number of entries) will influence how much memory is used. If merging together many small FlowFiles, a two-stage approach may be necessary in order to avoid excessive use of memory.") @UseCase(description="Concatenate FlowFiles with textual content together in order to create fewer, larger FlowFiles.",keywords={"concatenate","bundle","aggregate","bin","merge","combine","smash"},configuration="\"Merge Strategy\" = \"Bin Packing Algorithm\"\n\"Merge Format\" = \"Binary Concatenation\"\n\"Delimiter Strategy\" = \"Text\"\n\"Demarcator\" = \"\\n\" (a newline can be inserted by pressing Shift + Enter)\n\"Minimum Number of Entries\" = \"1\"\n\"Maximum Number of Entries\" = \"500000000\"\n\"Minimum Group Size\" = the minimum amount of data to write to an output FlowFile. A reasonable value might be \"128 MB\"\n\"Maximum Group Size\" = the maximum amount of data to write to an output FlowFile. A reasonable value might be \"256 MB\"\n\"Max Bin Age\" = the maximum amount of time to wait for incoming data before timing out and transferring the FlowFile along even though it is smaller than the Max Bin Age. A reasonable value might be \"5 mins\"\n") @UseCase(description="Concatenate FlowFiles with binary content together in order to create fewer, larger FlowFiles.",notes="Not all binary data can be concatenated together. Whether or not this configuration is valid depends on the type of your data.",keywords={"concatenate","bundle","aggregate","bin","merge","combine","smash"},configuration="\"Merge Strategy\" = \"Bin Packing Algorithm\"\n\"Merge Format\" = \"Binary Concatenation\"\n\"Delimiter Strategy\" = \"Text\"\n\"Minimum Number of Entries\" = \"1\"\n\"Maximum Number of Entries\" = \"500000000\"\n\"Minimum Group Size\" = the minimum amount of data to write to an output FlowFile. A reasonable value might be \"128 MB\"\n\"Maximum Group Size\" = the maximum amount of data to write to an output FlowFile. A reasonable value might be \"256 MB\"\n\"Max Bin Age\" = the maximum amount of time to wait for incoming data before timing out and transferring the FlowFile along even though it is smaller than the Max Bin Age. A reasonable value might be \"5 mins\"\n") @UseCase(description="Reassemble a FlowFile that was previously split apart into smaller FlowFiles by a processor such as SplitText, UnpackContext, SplitRecord, etc.",keywords={"reassemble","repack","merge","recombine"},configuration="\"Merge Strategy\" = \"Defragment\"\n\"Merge Format\" = the value of Merge Format depends on the desired output format. If the file was previously zipped together and was split apart by UnpackContent,\n a Merge Format of \"ZIP\" makes sense. If it was previously a .tar file, a Merge Format of \"TAR\" makes sense. If the data is textual, \"Binary Concatenation\" can be\n used to combine the text into a single document.\n\"Delimiter Strategy\" = \"Text\"\n\"Max Bin Age\" = the maximum amount of time to wait for incoming data before timing out and transferring the fragments to \'failure\'. A reasonable value might be \"5 mins\"\n\nFor textual data, \"Demarcator\" should be set to a newline (\\n), set by pressing Shift+Enter in the UI. For binary data, \"Demarcator\" should be left blank.\n") public class MergeContent extends BinFiles

  • Field Details

    • FRAGMENT_ID_ATTRIBUTE

      public static final String FRAGMENT_ID_ATTRIBUTE

    • FRAGMENT_INDEX_ATTRIBUTE

      public static final String FRAGMENT_INDEX_ATTRIBUTE

    • FRAGMENT_COUNT_ATTRIBUTE

      public static final String FRAGMENT_COUNT_ATTRIBUTE

    • SEGMENT_ORIGINAL_FILENAME

      public static final String SEGMENT_ORIGINAL_FILENAME

    • TAR_PERMISSIONS_ATTRIBUTE

      public static final String TAR_PERMISSIONS_ATTRIBUTE
      See Also:

    • MERGE_COUNT_ATTRIBUTE

      public static final String MERGE_COUNT_ATTRIBUTE
      See Also:

    • MERGE_BIN_AGE_ATTRIBUTE

      public static final String MERGE_BIN_AGE_ATTRIBUTE
      See Also:

    • MERGE_UUID_ATTRIBUTE

      public static final String MERGE_UUID_ATTRIBUTE
      See Also:

    • REASON_FOR_MERGING

      public static final String REASON_FOR_MERGING
      See Also:

    • MERGE_STRATEGY

      public static final org.apache.nifi.components.PropertyDescriptor MERGE_STRATEGY

    • MERGE_FORMAT

      public static final org.apache.nifi.components.PropertyDescriptor MERGE_FORMAT

    • METADATA_STRATEGY

      public static final org.apache.nifi.components.PropertyDescriptor METADATA_STRATEGY

    • CORRELATION_ATTRIBUTE_NAME

      public static final org.apache.nifi.components.PropertyDescriptor CORRELATION_ATTRIBUTE_NAME

    • DELIMITER_STRATEGY

      public static final org.apache.nifi.components.PropertyDescriptor DELIMITER_STRATEGY

    • DEMARCATOR

      public static final org.apache.nifi.components.PropertyDescriptor DEMARCATOR

    • COMPRESSION_LEVEL

      public static final org.apache.nifi.components.PropertyDescriptor COMPRESSION_LEVEL

    • KEEP_PATH

      public static final org.apache.nifi.components.PropertyDescriptor KEEP_PATH

    • TAR_MODIFIED_TIME

      public static final org.apache.nifi.components.PropertyDescriptor TAR_MODIFIED_TIME

    • BIN_TERMINATION_CHECK

      public static final org.apache.nifi.components.PropertyDescriptor BIN_TERMINATION_CHECK

    • FLOWFILE_INSERTION_STRATEGY

      public static final org.apache.nifi.components.PropertyDescriptor FLOWFILE_INSERTION_STRATEGY

    • PROPERTY_DESCRIPTORS

      private static final List<org.apache.nifi.components.PropertyDescriptor> PROPERTY_DESCRIPTORS

    • REL_MERGED

      public static final org.apache.nifi.processor.Relationship REL_MERGED

    • RELATIONSHIPS

      private static final Set<org.apache.nifi.processor.Relationship> RELATIONSHIPS

    • NUMBER_PATTERN

      public static final Pattern NUMBER_PATTERN

  • Constructor Details

    • MergeContent

      public MergeContent()

  • Method Details

    • addBinPackingDependency

      private static org.apache.nifi.components.PropertyDescriptor addBinPackingDependency(org.apache.nifi.components.PropertyDescriptor original)

    • getRelationships

      public Set<org.apache.nifi.processor.Relationship> getRelationships()
      Specified by:
      getRelationships in interface org.apache.nifi.processor.Processor
      Overrides:
      getRelationships in class org.apache.nifi.processor.AbstractSessionFactoryProcessor

    • getSupportedPropertyDescriptors

      protected List<org.apache.nifi.components.PropertyDescriptor> getSupportedPropertyDescriptors()
      Overrides:
      getSupportedPropertyDescriptors in class org.apache.nifi.components.AbstractConfigurableComponent

    • additionalCustomValidation

      protected Collection<org.apache.nifi.components.ValidationResult> additionalCustomValidation(org.apache.nifi.components.ValidationContext context)
      Overrides:
      additionalCustomValidation in class BinFiles

    • readContent

      private byte[] readContent(String filename) throws IOException
      Throws:
      IOException

    • preprocessFlowFile

      protected org.apache.nifi.flowfile.FlowFile preprocessFlowFile(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile)
      Specified by:
      preprocessFlowFile in class BinFiles

    • getGroupId

      protected String getGroupId(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.flowfile.FlowFile flowFile, org.apache.nifi.processor.ProcessSession session)
      Specified by:
      getGroupId in class BinFiles

    • setUpBinManager

      protected void setUpBinManager(BinManager binManager, org.apache.nifi.processor.ProcessContext context)
      Specified by:
      setUpBinManager in class BinFiles

    • processBin

      protected BinProcessingResult processBin(Bin bin, org.apache.nifi.processor.ProcessContext context) throws org.apache.nifi.processor.exception.ProcessException
      Specified by:
      processBin in class BinFiles
      Throws:
      org.apache.nifi.processor.exception.ProcessException

    • getMinEntries

      protected int getMinEntries(org.apache.nifi.context.PropertyContext context)
      Overrides:
      getMinEntries in class BinFiles

    • getMaxEntries

      protected int getMaxEntries(org.apache.nifi.context.PropertyContext context)
      Overrides:
      getMaxEntries in class BinFiles

    • getMinBytes

      protected long getMinBytes(org.apache.nifi.context.PropertyContext context)
      Overrides:
      getMinBytes in class BinFiles

    • getMaxBytes

      protected long getMaxBytes(org.apache.nifi.context.PropertyContext context)
      Overrides:
      getMaxBytes in class BinFiles

    • getDefragmentValidationError

      private String getDefragmentValidationError(List<org.apache.nifi.flowfile.FlowFile> binContents)

    • isNumber

      private boolean isNumber(String value)

    • removeFlowFileFromSession

      private void removeFlowFileFromSession(org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, org.apache.nifi.processor.ProcessContext context)

    • getPath

      private String getPath(org.apache.nifi.flowfile.FlowFile flowFile)

    • createFilename

      private String createFilename(List<org.apache.nifi.flowfile.FlowFile> flowFiles)