QueryDatabaseTableRecord

Description

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.

Tags

database, jdbc, query, record, select, sql

Properties

In the list below required Properties are shown with an asterisk (*). Other properties are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.

Display Name	API Name	Default Value	Allowable Values	Description
Database Connection Pooling Service *	Database Connection Pooling Service		Controller Service: DBCPService Implementations: DBCPConnectionPoolLookup DBCPConnectionPool HikariCPConnectionPool SnowflakeConnectionService	The Controller Service that is used to obtain a connection to the database.
Database Type *	db-fetch-db-type	Generic	Generic Oracle Oracle 12+ MS SQL 2012+ MS SQL 2008 MySQL PostgreSQL Phoenix	The type/flavor of database, used for generating database-specific code. In many cases the Generic type should suffice, but some databases (such as Oracle) require custom SQL clauses.
Table Name *	Table Name			The name of the database table to be queried. When a custom query is used, this property is used to alias the query and appears as an attribute on the FlowFile. Supports Expression Language, using FlowFile attributes and Environment variables.
Columns to Return	Columns to Return			A comma-separated list of column names to be used in the query. If your database requires special treatment of the names (quoting, e.g.), each name should include such treatment. If no column names are supplied, all columns in the specified table will be returned. NOTE: It is important to use consistent column names for a given table for incremental fetch to work properly. Supports Expression Language, using FlowFile attributes and Environment variables.
Additional WHERE clause	db-fetch-where-clause			A custom clause to be added in the WHERE condition when building SQL queries. Supports Expression Language, using FlowFile attributes and Environment variables.
Custom Query	db-fetch-sql-query			A custom SQL query used to retrieve data. Instead of building a SQL query from other properties, this query will be wrapped as a sub-query. Query must have no ORDER BY statement. Supports Expression Language, using Environment variables.
Record Writer *	qdbtr-record-writer		Controller Service: RecordSetWriterFactory Implementations: AvroRecordSetWriter CSVRecordSetWriter FreeFormTextRecordSetWriter JsonRecordSetWriter RecordSetWriterLookup ScriptedRecordSetWriter XMLRecordSetWriter	Specifies the Controller Service to use for writing results to a FlowFile. The Record Writer may use Inherit Schema to emulate the inferred schema behavior, i.e. an explicit schema need not be defined in the writer, and will be supplied by the same logic used to infer the schema from the column types.
Maximum-value Columns	Maximum-value Columns			A comma-separated list of column names. The processor will keep track of the maximum value for each column that has been returned since the processor started running. Using multiple columns implies an order to the column list, and each column's values are expected to increase more slowly than the previous columns' values. Thus, using multiple columns implies a hierarchical structure of columns, which is usually used for partitioning tables. This processor can be used to retrieve only those rows that have been added/updated since the last retrieval. Note that some JDBC types such as bit/boolean are not conducive to maintaining maximum value, so columns of these types should not be listed in this property, and will result in error(s) during processing. If no columns are provided, all rows from the table will be considered, which could have a performance impact. NOTE: It is important to use consistent max-value column names for a given table for incremental fetch to work properly. Supports Expression Language, using FlowFile attributes and Environment variables.
Initial Load Strategy *	initial-load-strategy	Start at Beginning	Start at Beginning Start at Current Maximum Values	How to handle existing rows in the database table when the processor is started for the first time (or its state has been cleared). The property will be ignored, if any 'initial.maxvalue.*' dynamic property has also been configured.
Max Wait Time *	Max Wait Time	0 seconds		The maximum amount of time allowed for a running SQL select query , zero means there is no limit. Max time less than 1 second will be equal to zero. Supports Expression Language, using FlowFile attributes and Environment variables.
Fetch Size *	Fetch Size	0		The number of result rows to be fetched from the result set at a time. This is a hint to the database driver and may not be honored and/or exact. If the value specified is zero, then the hint is ignored. If using PostgreSQL, then 'Set Auto Commit' must be equal to 'false' to cause 'Fetch Size' to take effect. Supports Expression Language, using Environment variables.
Set Auto Commit	Set Auto Commit		true false	Allows enabling or disabling the auto commit functionality of the DB connection. Default value is 'No value set'. 'No value set' will leave the db connection's auto commit mode unchanged. For some JDBC drivers such as PostgreSQL driver, it is required to disable the auto commit functionality to get the 'Fetch Size' setting to take effect. When auto commit is enabled, PostgreSQL driver ignores 'Fetch Size' setting and loads all rows of the result set to memory at once. This could lead for a large amount of memory usage when executing queries which fetch large data sets. More Details of this behaviour in PostgreSQL driver can be found in https://jdbc.postgresql.org//documentation/head/query.html. Supports Expression Language, using Environment variables.
Max Rows Per Flow File *	qdbt-max-rows	0		The maximum number of result rows that will be included in a single FlowFile. This will allow you to break up very large result sets into multiple FlowFiles. If the value specified is zero, then all rows are returned in a single FlowFile. Supports Expression Language, using Environment variables.
Output Batch Size *	qdbt-output-batch-size	0		The number of output FlowFiles to queue before committing the process session. When set to zero, the session will be committed when all result set rows have been processed and the output FlowFiles are ready for transfer to the downstream relationship. For large result sets, this can cause a large burst of FlowFiles to be transferred at the end of processor execution. If this property is set, then when the specified number of FlowFiles are ready for transfer, then the session will be committed, thus releasing the FlowFiles to the downstream relationship. NOTE: The maxvalue.* and fragment.count attributes will not be set on FlowFiles when this property is set. Supports Expression Language, using Environment variables.
Maximum Number of Fragments *	qdbt-max-frags	0		The maximum number of fragments. If the value specified is zero, then all fragments are returned. This prevents OutOfMemoryError when this processor ingests huge table. NOTE: Setting this property can result in data loss, as the incoming results are not ordered, and fragments may end at arbitrary boundaries where rows are not included in the result set. Supports Expression Language, using Environment variables.
Normalize Table/Column Names *	qdbtr-normalize	false	true false	Whether to change characters in column names when creating the output schema. For example, colons and periods will be changed to underscores.
Use Avro Logical Types *	dbf-user-logical-types	false	true false	Whether to use Avro Logical Types for DECIMAL/NUMBER, DATE, TIME and TIMESTAMP columns. If disabled, written as string. If enabled, Logical types are used and written as its underlying type, specifically, DECIMAL/NUMBER as logical 'decimal': written as bytes with additional precision and scale meta data, DATE as logical 'date-millis': written as int denoting days since Unix epoch (1970-01-01), TIME as logical 'time-millis': written as int denoting milliseconds since Unix epoch, and TIMESTAMP as logical 'timestamp-millis': written as long denoting milliseconds since Unix epoch. If a reader of written Avro records also knows these logical types, then these values can be deserialized with more context depending on reader implementation.
Default Decimal Precision *	dbf-default-precision	10		When a DECIMAL/NUMBER value is written as a 'decimal' Avro logical type, a specific 'precision' denoting number of available digits is required. Generally, precision is defined by column data type definition or database engines default. However undefined precision (0) can be returned from some database engines. 'Default Decimal Precision' is used when writing those undefined precision numbers. Supports Expression Language, using Environment variables.
Default Decimal Scale *	dbf-default-scale	0		When a DECIMAL/NUMBER value is written as a 'decimal' Avro logical type, a specific 'scale' denoting number of available decimal digits is required. Generally, scale is defined by column data type definition or database engines default. However when undefined precision (0) is returned, scale can also be uncertain with some database engines. 'Default Decimal Scale' is used when writing those undefined numbers. If a value has more decimals than specified scale, then the value will be rounded-up, e.g. 1.53 becomes 2 with scale 0, and 1.5 with scale 1. Supports Expression Language, using Environment variables.

Dynamic Properties

Name	Value	Description
initial.maxvalue.<max_value_column>	Initial maximum value for the specified column	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). Supports Expression Language: Yes, evaluated using Environment variables.

Relationships

Name	Description
success	Successfully created FlowFile from SQL query result set.

Reads Attributes

This processor does not read attributes.

Writes Attributes

Name	Description
fragment.count	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.
fragment.identifier	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.
fragment.index	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
maxvalue.*	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.
mime.type	Sets the mime.type attribute to the MIME Type specified by the Record Writer.
querydbtable.row.count	The number of rows selected by the query
record.count	The number of records output by the Record Writer.
tablename	Name of the table being queried

State Management

Scope

Description

CLUSTER

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

Restricted

This component is not restricted.

Input Requirement

This component does not allow an incoming relationship.

Example Use Cases

Use Case 1

Retrieve all rows from a database table.

Configuration

Configure the "Database Connection Pooling Service" to specify a Connection Pooling Service so that the Processor knows how to connect to the database.
Set the "Database Type" property to the type of database to query, or "Generic" if the database vendor is not listed.
Set the "Table Name" property to the name of the table to retrieve records from.
Configure the "Record Writer" to specify a Record Writer that is appropriate for the desired output format.
Set 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
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.
Set the "Initial Load Strategy" property to "Start at Beginning".
Set 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.
Set the "Max Rows Per Flow File" to a value that allows efficient processing, such as 1000 or 10000.
Set 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.
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
memory utilization.

Use Case 2

Perform an incremental load of a single database table, fetching only new rows as they are added to the table.

Configuration

Configure the "Database Connection Pooling Service" to specify a Connection Pooling Service so that the Processor knows how to connect to the database.
Set the "Database Type" property to the type of database to query, or "Generic" if the database vendor is not listed.
Set the "Table Name" property to the name of the table to retrieve records from.
Configure the "Record Writer" to specify a Record Writer that is appropriate for the desired output format.
Set 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
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.
Set the "Initial Load Strategy" property to "Start at Current Maximum Values".
Set 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.
Set the "Max Rows Per Flow File" to a value that allows efficient processing, such as 1000 or 10000.
Set 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.
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
memory utilization.

Example Use Cases Involving Other Components

Multiprocessor Use Case 1

Perform an incremental load of multiple database tables, fetching only new rows as they are added to the tables.

Components Involved

• ListDatabaseTables
  1. Configure the "Database Connection Pooling Service" property to specify a Connection Pool that is applicable for interacting with your database.
  2. Set the "Catalog" property to the name of the database Catalog;
  3. set the "Schema Pattern" property to a Java Regular Expression that matches all database Schemas that should be included; and
  4. set the "Table Name Pattern" property to a Java Regular Expression that matches the names of all tables that should be included.
  5. In order to perform an incremental load of all tables, leave the Catalog, Schema Pattern, and Table Name Pattern unset.
  6. Leave the RecordWriter property unset.
  7. Connect the 'success' relationship to QueryDatabaseTableRecord.
• QueryDatabaseTableRecord
  1. Configure the "Database Connection Pooling Service" to the same Connection Pool that was used in ListDatabaseTables.
  2. Set the "Database Type" property to the type of database to query, or "Generic" if the database vendor is not listed.
  3. Set the "Table Name" property to "${db.table.fullname}"
  4. Configure the "Record Writer" to specify a Record Writer that is appropriate for the desired output format.
  5. Set 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
  6. 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.
  7. Set the "Initial Load Strategy" property to "Start at Current Maximum Values".
  8. Set 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.
  9. Set the "Max Rows Per Flow File" to a value that allows efficient processing, such as 1000 or 10000.
  10. Set 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.
  11. 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
  12. memory utilization.

System Resource Considerations

This component does not specify system resource considerations.

See Also

ExecuteSQL, GenerateTableFetch