Package org.apache.nifi.processors.standard

Class MergeRecord

java.lang.Object
org.apache.nifi.components.AbstractConfigurableComponent
org.apache.nifi.processor.AbstractSessionFactoryProcessor
org.apache.nifi.processors.standard.MergeRecord
All Implemented Interfaces:
org.apache.nifi.components.ConfigurableComponent, org.apache.nifi.processor.Processor
@SideEffectFree @TriggerWhenEmpty @InputRequirement(INPUT_REQUIRED) @Tags({"merge","record","content","correlation","stream","event"}) @CapabilityDescription("This Processor merges together multiple record-oriented FlowFiles into a single FlowFile that contains all of the Records of the input FlowFiles. This Processor works by creating \'bins\' and then adding FlowFiles to these bins until they are full. Once a bin is full, all of the FlowFiles will be combined into a single output FlowFile, and that FlowFile will be routed to the \'merged\' Relationship. A bin will consist of potentially many \'like FlowFiles\'. In order for two FlowFiles to be considered \'like FlowFiles\', they must have the same Schema (as identified by the Record Reader) and, if the <Correlation Attribute Name> property is set, the same value for the specified attribute. See Processor Usage and Additional Details for more information. 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.count",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 indicates how many FlowFiles should be expected in the given bundle.") @WritesAttribute(attribute="record.count",description="The merged FlowFile will have a \'record.count\' attribute indicating the number of records that were written to the FlowFile.") @WritesAttribute(attribute="mime.type",description="The MIME Type indicated by the Record Writer") @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 FlowFile that will be added to the original FlowFiles attributes") @WritesAttribute(attribute="merge.completion.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.") @WritesAttribute(attribute="<Attributes from Record Writer>",description="Any Attribute that the configured Record Writer returns will be added to the FlowFile.") @SeeAlso({MergeContent.class,SplitRecord.class,PartitionRecord.class}) @UseCase(description="Combine together many arbitrary Records in order to create a single, larger file", configuration="Configure the \"Record Reader\" to specify a Record Reader that is appropriate for the incoming data type.\nConfigure the \"Record Writer\" to specify a Record Writer that is appropriate for the desired output data type.\nSet \"Merge Strategy\" to `Bin-Packing Algorithm`.\nSet the \"Minimum Bin Size\" to desired file size of the merged output file. For example, a value of `1 MB` will result in not merging data until at least\n 1 MB of data is available (unless the Max Bin Age is reached first). If there is no desired minimum file size, leave the default value of `0 B`.\nSet the \"Minimum Number of Records\" property to the minimum number of Records that should be included in the merged output file. For example, setting the value\n to `10000` ensures that the output file will have at least 10,000 Records in it (unless the Max Bin Age is reached first).\nSet the \"Max Bin Age\" to specify the maximum amount of time to hold data before merging. This can be thought of as a \"timeout\" at which time the Processor will\n merge whatever data it is, even if the \"Minimum Bin Size\" and \"Minimum Number of Records\" has not been reached. It is always recommended to set the value.\n A reasonable default might be `10 mins` if there is no other latency requirement.\n\nConnect the \'merged\' Relationship to the next component in the flow. Auto-terminate the \'original\' Relationship.\n") @MultiProcessorUseCase(description="Combine together many Records that have the same value for a particular field in the data, in order to create a single, larger file",keywords={"merge","combine","aggregate","like records","similar data"},configurations={@ProcessorConfiguration(processorClass=PartitionRecord.class,configuration="Configure the \"Record Reader\" to specify a Record Reader that is appropriate for the incoming data type.\nConfigure the \"Record Writer\" to specify a Record Writer that is appropriate for the desired output data type.\n\nAdd a single additional property. The name of the property should describe the field on which the data is being merged together.\nThe property\'s value should be a RecordPath that specifies which output FlowFile the Record belongs to.\n\nFor example, to merge together data that has the same value for the \"productSku\" field, add a property named `productSku` with a value of `/productSku`.\n\nConnect the \"success\" Relationship to MergeRecord.\nAuto-terminate the \"original\" Relationship.\n"),@ProcessorConfiguration(processorClass=MergeRecord.class,configuration="Configure the \"Record Reader\" to specify a Record Reader that is appropriate for the incoming data type.\nConfigure the \"Record Writer\" to specify a Record Writer that is appropriate for the desired output data type.\nSet \"Merge Strategy\" to `Bin-Packing Algorithm`.\nSet the \"Minimum Bin Size\" to desired file size of the merged output file. For example, a value of `1 MB` will result in not merging data until at least\n 1 MB of data is available (unless the Max Bin Age is reached first). If there is no desired minimum file size, leave the default value of `0 B`.\nSet the \"Minimum Number of Records\" property to the minimum number of Records that should be included in the merged output file. For example, setting the value\n to `10000` ensures that the output file will have at least 10,000 Records in it (unless the Max Bin Age is reached first).\nSet the \"Maximum Number of Records\" property to a value at least as large as the \"Minimum Number of Records.\" If there is no need to limit the maximum number of\n records per file, this number can be set to a value that will never be reached such as `1000000000`.\nSet the \"Max Bin Age\" to specify the maximum amount of time to hold data before merging. This can be thought of as a \"timeout\" at which time the Processor will\n merge whatever data it is, even if the \"Minimum Bin Size\" and \"Minimum Number of Records\" has not been reached. It is always recommended to set the value.\n A reasonable default might be `10 mins` if there is no other latency requirement.\nSet the value of the \"Correlation Attribute Name\" property to the name of the property that you added in the PartitionRecord Processor. For example, if merging data\n based on the \"productSku\" field, the property in PartitionRecord was named `productSku` so the value of the \"Correlation Attribute Name\" property should\n be `productSku`.\nSet the \"Maximum Number of Bins\" property to a value that is at least as large as the different number of values that will be present for the Correlation Attribute.\n For example, if you expect 1,000 different SKUs, set this value to at least `1001`. It is not advisable, though, to set the value above 10,000.\n\nConnect the \'merged\' Relationship to the next component in the flow.\nAuto-terminate the \'original\' Relationship.\n")}) public class MergeRecord extends org.apache.nifi.processor.AbstractSessionFactoryProcessor

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    private final AtomicReference<RecordBinManager>
    binManager
     
    static final org.apache.nifi.components.PropertyDescriptor
    CORRELATION_ATTRIBUTE_NAME
     
    static final String
    FRAGMENT_COUNT_ATTRIBUTE
     
    static final String
    FRAGMENT_ID_ATTRIBUTE
     
    static final String
    FRAGMENT_INDEX_ATTRIBUTE
     
    static final org.apache.nifi.components.PropertyDescriptor
    MAX_BIN_AGE
     
    static final org.apache.nifi.components.PropertyDescriptor
    MAX_BIN_COUNT
     
    static final org.apache.nifi.components.PropertyDescriptor
    MAX_RECORDS
     
    static final org.apache.nifi.components.PropertyDescriptor
    MAX_SIZE
     
    static final String
    MERGE_BIN_AGE_ATTRIBUTE
     
    static final String
    MERGE_COMPLETION_REASON
     
    static final String
    MERGE_COUNT_ATTRIBUTE
     
    static final org.apache.nifi.components.PropertyDescriptor
    MERGE_STRATEGY
     
    static final org.apache.nifi.components.AllowableValue
    MERGE_STRATEGY_BIN_PACK
     
    static final org.apache.nifi.components.AllowableValue
    MERGE_STRATEGY_DEFRAGMENT
     
    static final String
    MERGE_UUID_ATTRIBUTE
     
    static final org.apache.nifi.components.PropertyDescriptor
    MIN_RECORDS
     
    static final org.apache.nifi.components.PropertyDescriptor
    MIN_SIZE
     
    private static final List<org.apache.nifi.components.PropertyDescriptor>
    PROPERTY_DESCRIPTORS
     
    static final org.apache.nifi.components.PropertyDescriptor
    RECORD_READER
     
    static final org.apache.nifi.components.PropertyDescriptor
    RECORD_WRITER
     
    static final org.apache.nifi.processor.Relationship
    REL_FAILURE
     
    static final org.apache.nifi.processor.Relationship
    REL_MERGED
     
    static final org.apache.nifi.processor.Relationship
    REL_ORIGINAL
     
    private static final Set<org.apache.nifi.processor.Relationship>
    RELATIONSHIPS
     

  • Constructor Summary

    Constructors
    Constructor
    Description
    MergeRecord()
     

  • Method Summary

    Modifier and Type
    Method
    Description
    private void
    binFlowFile(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.flowfile.FlowFile flowFile, org.apache.nifi.processor.ProcessSession session, RecordBinManager binManager, boolean block)
     
    protected Collection<org.apache.nifi.components.ValidationResult>
    customValidate(org.apache.nifi.components.ValidationContext validationContext)
     
    (package private) int
    getBinCount()
     
    protected String
    getGroupId(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.flowfile.FlowFile flowFile, RecordSchema schema, org.apache.nifi.processor.ProcessSession session)
     
    Set<org.apache.nifi.processor.Relationship>
    getRelationships()
     
    protected List<org.apache.nifi.components.PropertyDescriptor>
    getSupportedPropertyDescriptors()
     
    void
    onTrigger(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSessionFactory sessionFactory)
     
    final void
    resetState()
     

    Methods inherited from class org.apache.nifi.processor.AbstractSessionFactoryProcessor

    getControllerServiceLookup, getIdentifier, getLogger, getNodeTypeProvider, init, initialize, isConfigurationRestored, isScheduled, toString, updateConfiguredRestoredTrue, updateScheduledFalse, updateScheduledTrue

    Methods inherited from class org.apache.nifi.components.AbstractConfigurableComponent

    equals, getPropertyDescriptor, getPropertyDescriptors, getSupportedDynamicPropertyDescriptor, hashCode, onPropertyModified, validate

    Methods inherited from class java.lang.Object

    clone, finalize, getClass, notify, notifyAll, wait, wait, wait

    Methods inherited from interface org.apache.nifi.components.ConfigurableComponent

    getPropertyDescriptor, getPropertyDescriptors, onPropertyModified, validate

    Methods inherited from interface org.apache.nifi.processor.Processor

    isStateful, migrateProperties, migrateRelationships

  • 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

    • 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:

    • MERGE_COMPLETION_REASON

      public static final String MERGE_COMPLETION_REASON
      See Also:

    • MERGE_STRATEGY_BIN_PACK

      public static final org.apache.nifi.components.AllowableValue MERGE_STRATEGY_BIN_PACK

    • MERGE_STRATEGY_DEFRAGMENT

      public static final org.apache.nifi.components.AllowableValue MERGE_STRATEGY_DEFRAGMENT

    • RECORD_READER

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

    • RECORD_WRITER

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

    • MERGE_STRATEGY

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

    • CORRELATION_ATTRIBUTE_NAME

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

    • MIN_SIZE

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

    • MAX_SIZE

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

    • MIN_RECORDS

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

    • MAX_RECORDS

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

    • MAX_BIN_COUNT

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

    • MAX_BIN_AGE

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

    • 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

    • REL_ORIGINAL

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

    • REL_FAILURE

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

    • RELATIONSHIPS

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

    • binManager

      private final AtomicReference<RecordBinManager> binManager

  • Constructor Details

    • MergeRecord

      public MergeRecord()

  • Method Details

    • getSupportedPropertyDescriptors

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

    • 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

    • resetState

      @OnStopped public final void resetState()

    • customValidate

      protected Collection<org.apache.nifi.components.ValidationResult> customValidate(org.apache.nifi.components.ValidationContext validationContext)
      Overrides:
      customValidate in class org.apache.nifi.components.AbstractConfigurableComponent

    • onTrigger

      public void onTrigger(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSessionFactory sessionFactory) throws org.apache.nifi.processor.exception.ProcessException
      Throws:
      org.apache.nifi.processor.exception.ProcessException

    • binFlowFile

      private void binFlowFile(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.flowfile.FlowFile flowFile, org.apache.nifi.processor.ProcessSession session, RecordBinManager binManager, boolean block)

    • getGroupId

      protected String getGroupId(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.flowfile.FlowFile flowFile, RecordSchema schema, org.apache.nifi.processor.ProcessSession session)

    • getBinCount

      int getBinCount()