Package org.apache.nifi.processors.standard

Class QueryDatabaseTableRecord

java.lang.Object
org.apache.nifi.components.AbstractConfigurableComponent
org.apache.nifi.processor.AbstractSessionFactoryProcessor
org.apache.nifi.processors.standard.AbstractDatabaseFetchProcessor
org.apache.nifi.processors.standard.AbstractQueryDatabaseTable
org.apache.nifi.processors.standard.QueryDatabaseTableRecord
All Implemented Interfaces:
org.apache.nifi.components.ConfigurableComponent, org.apache.nifi.processor.Processor
@TriggerSerially @InputRequirement(INPUT_FORBIDDEN) @Tags({"sql","select","jdbc","query","database","record"}) @SeeAlso({GenerateTableFetch.class,ExecuteSQL.class}) @CapabilityDescription("Generates a SQL select query, or uses a provided statement, and executes it to fetch all rows whose values in the specified Maximum Value column(s) are larger than the previously-seen maxima. Query result will be converted to the format specified by the record writer. Expression Language is supported for several properties, but no incoming connections are permitted. The Environment/System properties may be used to provide values for any property containing Expression Language. If it is desired to leverage flow file attributes to perform these queries, the GenerateTableFetch and/or ExecuteSQL processors can be used for this purpose. Streaming is used so arbitrarily large result sets are supported. This processor can be scheduled to run on a timer or cron expression, using the standard scheduling methods. This processor is intended to be run on the Primary Node only. FlowFile attribute \'querydbtable.row.count\' indicates how many rows were selected.") @Stateful(scopes=CLUSTER, description="After performing a query on the specified table, the maximum values for the specified column(s) will be retained for use in future executions of the query. This allows the Processor to fetch only those records that have max values greater than the retained values. This can be used for incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor per the State Management documentation") @WritesAttribute(attribute="tablename",description="Name of the table being queried") @WritesAttribute(attribute="querydbtable.row.count",description="The number of rows selected by the query") @WritesAttribute(attribute="fragment.identifier",description="If \'Max Rows Per Flow File\' is set then all FlowFiles from the same query result set will have the same value for the fragment.identifier attribute. This can then be used to correlate the results.") @WritesAttribute(attribute="fragment.count",description="If \'Max Rows Per Flow File\' is set then this is the total number of FlowFiles produced by a single ResultSet. This can be used in conjunction with the fragment.identifier attribute in order to know how many FlowFiles belonged to the same incoming ResultSet. If Output Batch Size is set, then this attribute will not be populated.") @WritesAttribute(attribute="fragment.index",description="If \'Max Rows Per Flow File\' is set then the position of this FlowFile in the list of outgoing FlowFiles that were all derived from the same result set FlowFile. This can be used in conjunction with the fragment.identifier attribute to know which FlowFiles originated from the same query result set and in what order FlowFiles were produced") @WritesAttribute(attribute="maxvalue.*",description="Each attribute contains the observed maximum value of a specified \'Maximum-value Column\'. The suffix of the attribute is the name of the column. If Output Batch Size is set, then this attribute will not be populated.") @WritesAttribute(attribute="mime.type",description="Sets the mime.type attribute to the MIME Type specified by the Record Writer.") @WritesAttribute(attribute="record.count",description="The number of records output by the Record Writer.") @DynamicProperty(name="initial.maxvalue.<max_value_column>", value="Initial maximum value for the specified column", expressionLanguageScope=ENVIRONMENT, description="Specifies an initial max value for max value column(s). Properties should be added in the format `initial.maxvalue.<max_value_column>`. This value is only used the first time the table is accessed (when a Maximum Value Column is specified).") @PrimaryNodeOnly @DefaultSchedule(strategy=TIMER_DRIVEN, period="1 min") @UseCase(description="Retrieve all rows from a database table.",keywords={"jdbc","rdbms","cdc","database","table","stream"},configuration="Configure the \"Database Connection Pooling Service\" to specify a Connection Pooling Service so that the Processor knows how to connect to the database.\nSet the \"Database Type\" property to the type of database to query, or \"Generic\" if the database vendor is not listed.\nSet the \"Table Name\" property to the name of the table to retrieve records from.\nConfigure the \"Record Writer\" to specify a Record Writer that is appropriate for the desired output format.\nSet the \"Maximum-value Columns\" property to a comma-separated list of columns whose values can be used to determine which values are new. For example, this might be set to\n an `id` column that is a one-up number, or a `last_modified` column that is a timestamp of when the row was last modified.\nSet the \"Initial Load Strategy\" property to \"Start at Beginning\".\nSet the \"Fetch Size\" to a number that avoids loading too much data into memory on the NiFi side. For example, a value of `1000` will load up to 1,000 rows of data.\nSet the \"Max Rows Per Flow File\" to a value that allows efficient processing, such as `1000` or `10000`.\nSet the \"Output Batch Size\" property to a value greater than `0`. A smaller value, such as `1` or even `20` will result in lower latency but also slightly lower throughput.\n A larger value such as `1000` will result in higher throughput but also higher latency. It is not recommended to set the value larger than `1000` as it can cause significant\n memory utilization.\n") @UseCase(description="Perform an incremental load of a single database table, fetching only new rows as they are added to the table.",keywords={"incremental load","rdbms","jdbc","cdc","database","table","stream"},configuration="Configure the \"Database Connection Pooling Service\" to specify a Connection Pooling Service so that the Processor knows how to connect to the database.\nSet the \"Database Type\" property to the type of database to query, or \"Generic\" if the database vendor is not listed.\nSet the \"Table Name\" property to the name of the table to retrieve records from.\nConfigure the \"Record Writer\" to specify a Record Writer that is appropriate for the desired output format.\nSet the \"Maximum-value Columns\" property to a comma-separated list of columns whose values can be used to determine which values are new. For example, this might be set to\n an `id` column that is a one-up number, or a `last_modified` column that is a timestamp of when the row was last modified.\nSet the \"Initial Load Strategy\" property to \"Start at Current Maximum Values\".\nSet the \"Fetch Size\" to a number that avoids loading too much data into memory on the NiFi side. For example, a value of `1000` will load up to 1,000 rows of data.\nSet the \"Max Rows Per Flow File\" to a value that allows efficient processing, such as `1000` or `10000`.\nSet the \"Output Batch Size\" property to a value greater than `0`. A smaller value, such as `1` or even `20` will result in lower latency but also slightly lower throughput.\n A larger value such as `1000` will result in higher throughput but also higher latency. It is not recommended to set the value larger than `1000` as it can cause significant\n memory utilization.\n") @MultiProcessorUseCase(description="Perform an incremental load of multiple database tables, fetching only new rows as they are added to the tables.",keywords={"incremental load","rdbms","jdbc","cdc","database","table","stream"},configurations={@ProcessorConfiguration(processorClass=ListDatabaseTables.class,configuration="Configure the \"Database Connection Pooling Service\" property to specify a Connection Pool that is applicable for interacting with your database.\n\nSet the \"Catalog\" property to the name of the database Catalog;\nset the \"Schema Pattern\" property to a Java Regular Expression that matches all database Schemas that should be included; and\nset the \"Table Name Pattern\" property to a Java Regular Expression that matches the names of all tables that should be included.\nIn order to perform an incremental load of all tables, leave the Catalog, Schema Pattern, and Table Name Pattern unset.\n\nLeave the RecordWriter property unset.\n\nConnect the \'success\' relationship to QueryDatabaseTableRecord.\n"),@ProcessorConfiguration(processorClass=QueryDatabaseTableRecord.class,configuration="Configure the \"Database Connection Pooling Service\" to the same Connection Pool that was used in ListDatabaseTables.\nSet the \"Database Type\" property to the type of database to query, or \"Generic\" if the database vendor is not listed.\nSet the \"Table Name\" property to \"${db.table.fullname}\"\nConfigure the \"Record Writer\" to specify a Record Writer that is appropriate for the desired output format.\nSet the \"Maximum-value Columns\" property to a comma-separated list of columns whose values can be used to determine which values are new. For example, this might be set to\n an `id` column that is a one-up number, or a `last_modified` column that is a timestamp of when the row was last modified.\nSet the \"Initial Load Strategy\" property to \"Start at Current Maximum Values\".\nSet the \"Fetch Size\" to a number that avoids loading too much data into memory on the NiFi side. For example, a value of `1000` will load up to 1,000 rows of data.\nSet the \"Max Rows Per Flow File\" to a value that allows efficient processing, such as `1000` or `10000`.\nSet the \"Output Batch Size\" property to a value greater than `0`. A smaller value, such as `1` or even `20` will result in lower latency but also slightly lower throughput.\n A larger value such as `1000` will result in higher throughput but also higher latency. It is not recommended to set the value larger than `1000` as it can cause significant\n memory utilization.\n")}) public class QueryDatabaseTableRecord extends AbstractQueryDatabaseTable

  • Field Details

    • TABLE_NAME

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

    • RECORD_WRITER_FACTORY

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

    • NORMALIZE_NAMES

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

    • PROPERTY_DESCRIPTORS

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

    • RELATIONSHIPS

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

  • Constructor Details

    • QueryDatabaseTableRecord

      public QueryDatabaseTableRecord()

  • Method Details