You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2023/11/16 01:28:26 UTC
(nifi) branch main updated: NIFI-12374: Added use case based documentation to explain how to perform incremental / full loads from databases
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new e1942a8346 NIFI-12374: Added use case based documentation to explain how to perform incremental / full loads from databases
e1942a8346 is described below
commit e1942a8346ee2d4133c9ee92425eef599e883082
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Wed Nov 15 14:20:20 2023 -0500
NIFI-12374: Added use case based documentation to explain how to perform incremental / full loads from databases
This closes #8030
Signed-off-by: David Handermann <ex...@apache.org>
---
.../processors/standard/ListDatabaseTables.java | 42 ++++++++++++++++
.../standard/QueryDatabaseTableRecord.java | 58 ++++++++++++++++++++++
2 files changed, 100 insertions(+)
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListDatabaseTables.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListDatabaseTables.java
index 998ecf1a71..56a001187e 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListDatabaseTables.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListDatabaseTables.java
@@ -24,6 +24,8 @@ import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.configuration.DefaultSchedule;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.MultiProcessorUseCase;
+import org.apache.nifi.annotation.documentation.ProcessorConfiguration;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
@@ -95,6 +97,46 @@ import java.util.stream.Stream;
+ "indicate that when the processor detects the interval has elapsed, the state will be reset and tables will be re-listed as a result. "
+ "This processor is meant to be run on the primary node only.")
@DefaultSchedule(strategy = SchedulingStrategy.TIMER_DRIVEN, period = "1 min")
+@MultiProcessorUseCase(
+ description="Perform a full load of a database, retrieving all rows from all tables, or a specific set of tables.",
+ keywords = {"full load", "rdbms", "jdbc", "database"},
+ 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.
+ Leave the RecordWriter property unset.
+
+ Set the "Catalog" property to the name of the database Catalog; leave it empty to include all catalogs.
+ Set the "Schema Pattern" property to a Java Regular Expression that matches all database Schemas that should be included; leave it empty to include all Schemas.
+ Set the "Table Name Pattern" property to a Java Regular Expression that matches the names of all tables that should be included; leave it empty to include all Tables.
+
+ Connect the "success" relationship to GenerateTableFetch.
+ """
+ ),
+ @ProcessorConfiguration(
+ processorClass = GenerateTableFetch.class,
+ configuration = """
+ Configure the "Database Connection Pooling Service" property to specify the same Connection Pool that was used in ListDatabaseTables.
+ Set the "Database Type" property to match the appropriate value for your RDBMS vendor.
+ Set "Table Name" to `${db.table.fullname}`
+ Leave the RecordWriter property unset.
+
+ Connect the "success" relationship to ExecuteSQLRecord.
+ """
+ ),
+ @ProcessorConfiguration(
+ processorClass = ExecuteSQLRecord.class,
+ configuration = """
+ Configure the "Database Connection Pooling Service" property to specify the same Connection Pool that was used in ListDatabaseTables.
+ Configure the "Record Writer" property to specify a Record Writer that is appropriate for the desired output data type.
+ Leave the "SQL select query" unset.
+
+ Connect the "success" relationship to the next Processor in the flow.
+ """
+ )
+ }
+)
public class ListDatabaseTables extends AbstractProcessor {
// Attribute names
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTableRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTableRecord.java
index 0d68e7e7de..5838d7e46c 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTableRecord.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTableRecord.java
@@ -26,6 +26,8 @@ import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.configuration.DefaultSchedule;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.MultiProcessorUseCase;
+import org.apache.nifi.annotation.documentation.ProcessorConfiguration;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.documentation.UseCase;
@@ -110,6 +112,62 @@ import static org.apache.nifi.util.db.JdbcProperties.VARIABLE_REGISTRY_ONLY_DEFA
memory utilization.
"""
)
+@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.
+ 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.
+ """
+)
+@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.
+
+ Set the "Catalog" property to the name of the database Catalog;
+ set the "Schema Pattern" property to a Java Regular Expression that matches all database Schemas that should be included; and
+ set the "Table Name Pattern" property to a Java Regular Expression that matches the names of all tables that should be included.
+ In order to perform an incremental load of all tables, leave the Catalog, Schema Pattern, and Table Name Pattern unset.
+
+ Leave the RecordWriter property unset.
+
+ Connect the 'success' relationship to QueryDatabaseTableRecord.
+ """
+ ),
+ @ProcessorConfiguration(
+ processorClass = QueryDatabaseTableRecord.class,
+ configuration = """
+ Configure the "Database Connection Pooling Service" to the same Connection Pool that was used in ListDatabaseTables.
+ 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 "${db.table.fullname}"
+ 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.
+ """
+ )
+ }
+)
public class QueryDatabaseTableRecord extends AbstractQueryDatabaseTable {
public static final PropertyDescriptor RECORD_WRITER_FACTORY = new PropertyDescriptor.Builder()