QueryDatabaseTableRecord

QueryDatabaseTableRecord Processor

类型全限定名：cn.org.acal.lynxus.processors.standard.QueryDatabaseTableRecord

所属 Bundle：cn.org.acal.lynxus:lynxus-standard-nar:2.11.8

描述

生成 SQL select 查询或使用提供的语句，并执行它以获取所有符合条件的行。

使用场景

从数据库表中检索所有行。

jdbcrdbmscdcdatabasetablestream

配置 "Database Connection Pooling Service" 以指定一个连接池服务，使处理器知道如何连接到数据库。
将 "Database Type" 属性设置为要查询的数据库类型，如果未列出数据库供应商则设置为 "Generic"。
将 "Table Name" 属性设置为要从中检索记录的表的名称。
配置 "Record Writer" 以指定适合所需输出格式的 Record Writer。
将 "Maximum-value Columns" 属性设置为一个逗号分隔的列列表，其值可用于确定哪些值是新的。例如，这可以设置为
    一个作为递增数字的 `id` 列，或一个作为该行上次修改时间戳的 `last_modified` 列。
将 "Initial Load Strategy" 属性设置为 "Start at Beginning"。
将 "Fetch Size" 设置为一个避免在 Lynxus 端将过多数据加载到内存中的数字。例如，值 `1000` 将加载最多 1,000 行数据。
将 "Max Rows Per FlowFile" 设置为允许高效处理的值，例如 `1000` 或 `10000`。
将 "Output Batch Size" 属性设置为大于 `0` 的值。较小的值（例如 `1` 甚至 `20`）将导致较低的延迟，但吞吐量也略低。
    较大的值（例如 `1000`）将导致较高的吞吐量，但延迟也较高。不建议将该值设置得大于 `1000`，因为它可能导致显著的
    内存占用。

对单个数据库表执行增量加载，仅获取添加到表中的新行。

incremental loadrdbmsjdbccdcdatabasetablestream

配置 "Database Connection Pooling Service" 以指定一个连接池服务，使处理器知道如何连接到数据库。
将 "Database Type" 属性设置为要查询的数据库类型，如果未列出数据库供应商则设置为 "Generic"。
将 "Table Name" 属性设置为要从中检索记录的表的名称。
配置 "Record Writer" 以指定适合所需输出格式的 Record Writer。
将 "Maximum-value Columns" 属性设置为一个逗号分隔的列列表，其值可用于确定哪些值是新的。例如，这可以设置为
    一个作为递增数字的 `id` 列，或一个作为该行上次修改时间戳的 `last_modified` 列。
将 "Initial Load Strategy" 属性设置为 "Start at Current Maximum Values"。
将 "Fetch Size" 设置为一个避免在 Lynxus 端将过多数据加载到内存中的数字。例如，值 `1000` 将加载最多 1,000 行数据。
将 "Max Rows Per FlowFile" 设置为允许高效处理的值，例如 `1000` 或 `10000`。
将 "Output Batch Size" 属性设置为大于 `0` 的值。较小的值（例如 `1` 甚至 `20`）将导致较低的延迟，但吞吐量也略低。
    较大的值（例如 `1000`）将导致较高的吞吐量，但延迟也较高。不建议将该值设置得大于 `1000`，因为它可能导致显著的
    内存占用。

多处理器协同场景

对多个数据库表执行增量加载，仅获取添加到表中的新行。

incremental loadrdbmsjdbccdcdatabasetablestream

配置 ListDatabaseTables：

配置 "Database Connection Pooling Service" 属性以指定适用于与你的数据库交互的连接池。

将 "Catalog" 属性设置为数据库 Catalog 的名称；
将 "Schema Pattern" 属性设置为一个匹配所有应包含的数据库 Schema 的 Java 正则表达式；并
将 "Table Name Pattern" 属性设置为一个匹配所有应包含的表名的 Java 正则表达式。
为了对所有表执行增量加载，请将 Catalog、Schema Pattern 和 Table Name Pattern 留空（不设置）。

将 RecordWriter 属性留空（不设置）。

将 'success' 关系连接到 QueryDatabaseTableRecord。

配置 QueryDatabaseTableRecord：

将 "Database Connection Pooling Service" 配置为与 ListDatabaseTables 中使用的相同的连接池。
将 "Database Type" 属性设置为要查询的数据库类型，如果未列出数据库供应商则设置为 "Generic"。
将 "Table Name" 属性设置为 "${db.table.fullname}"
配置 "Record Writer" 以指定适合所需输出格式的 Record Writer。
将 "Maximum-value Columns" 属性设置为一个逗号分隔的列列表，其值可用于确定哪些值是新的。例如，这可以设置为
    一个作为递增数字的 `id` 列，或一个作为该行上次修改时间戳的 `last_modified` 列。
将 "Initial Load Strategy" 属性设置为 "Start at Current Maximum Values"。
将 "Fetch Size" 设置为一个避免在 Lynxus 端将过多数据加载到内存中的数字。例如，值 `1000` 将加载最多 1,000 行数据。
将 "Max Rows Per FlowFile" 设置为允许高效处理的值，例如 `1000` 或 `10000`。
将 "Output Batch Size" 属性设置为大于 `0` 的值。较小的值（例如 `1` 甚至 `20`）将导致较低的延迟，但吞吐量也略低。
    较大的值（例如 `1000`）将导致较高的吞吐量，但延迟也较高。不建议将该值设置得大于 `1000`，因为它可能导致显著的
    内存占用。

属性

下表中，加粗的属性为必填项；其余为可选项。表格同时标注默认值、是否支持表达式语言，以及是否为敏感属性（值会被加密）。

名称	默认值	可选值	描述
Database Connection Pooling Service		控制器服务 API： `DBCPService` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	用于获取数据库连接的Controller Service。
数据库类型	`Generic`	Database Dialect Service — 需要为 SQL 语句配置 Database Dialect Service Generic — 生成 ANSI SQL Oracle — 生成符合 Oracle 规范的 SQL Oracle 12+ — 生成符合 Oracle 规范的 SQL（12 或更高版本） MS SQL 2012+ — 生成兼容 MS SQL 的 SQL（2012 或更高版本） MS SQL 2008 — 生成兼容 MS SQL 的 SQL（2008 版本） MySQL — 生成兼容 MySQL 的 SQL PostgreSQL — 生成兼容 PostgreSQL 的 SQL Phoenix — 生成符合 Phoenix 规范的 SQL	用于生成特定于服务或供应商的语句的数据库类型。通用类型支持大多数情况，但选择特定类型可以启用优化处理或额外功能。
Database Dialect Service		控制器服务 API： `DatabaseDialectService` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	数据库方言服务，用于生成特定于特定服务或供应商的语句。
Table Name			需要查询的数据库表名。当使用自定义查询时，此属性用于为查询指定别名，并显示为 FlowFile 上的属性。支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Columns to Return			逗号分隔的列名列表，用于查询中。如果您的数据库需要对名称进行特殊处理（例如引号），则每个名称应包含此类处理。如果未提供列名，将返回指定表中的所有列。注意：对于给定的表，使用一致的列名对于增量获取正常工作非常重要。支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Additional WHERE Clause			构建 SQL 查询时要在 WHERE 条件中添加的自定义子句。支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Custom Query			用于检索数据的自定义 SQL 查询。此查询将被包装为子查询，而不是从其他属性构建 SQL 查询。查询不能包含 ORDER BY 语句。支持表达式语言：是（仅使用环境/系统变量求值）
Record Writer		控制器服务 API： `RecordSetWriterFactory` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	指定用于将结果写入 FlowFile 的控制器服务。Record Writer 可以使用 Inherit Schema 来模拟推断的 Schema 行为，即不需要在 Writer 中定义显式 Schema，将使用与从列类型推断 Schema 相同的逻辑来提供。
Maximum-value Columns			逗号分隔的列名列表。处理器将跟踪自处理器启动以来返回的每列的最大值。使用多列意味着列列表具有顺序关系，每列的值预期比前一列的值增长更慢。因此，使用多列意味着列的分层结构，通常用于分区表。此处理器可用于仅检索自上次检索以来添加/更新的行。请注意，某些 JDBC 类型（如 bit/boolean）不利于维护最大值，因此不应在此属性中列出这些类型的列，否则会在处理过程中导致错误。如果未提供列，将考虑表中的所有行，这可能会影响性能。注意：对于给定的表，使用一致的最大值列名对于增量获取正常工作非常重要。支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
初始加载策略	`Start at Beginning`	从头开始 — 加载数据库表中所有现有行。从当前最大值开始 — 仅加载在 'Maximum-value Columns' 属性中配置的列的最大值之后新插入或更新的行。	处理器首次启动（或其状态已清除）时如何处理数据库表中的现有行。如果还配置了任何 'initial.maxvalue.*' 动态属性，则该属性将被忽略。
Max Wait Time	`0 seconds`		正在运行的 SQL 查询允许的最大时间，零表示没有限制。最大时间小于 1 秒将等于零。支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
获取大小	`0`		每次从结果集中获取的结果行数。这是对数据库驱动的提示，可能不会被遵守和/或精确。如果指定的值为零，则忽略该提示。如果使用 PostgreSQL，则 'Set Auto Commit' 必须等于 'false' 才能使 'Fetch Size' 生效。支持表达式语言：是（仅使用环境/系统变量求值）
设置自动提交		true false	允许启用或禁用数据库连接的自动提交功能。默认值为 'No value set'。'No value set' 将保持数据库连接的自动提交模式不变。对于某些 JDBC 驱动（如 PostgreSQL 驱动），需要禁用自动提交功能才能使 'Fetch Size' 设置生效。当启用自动提交时，PostgreSQL 驱动会忽略 'Fetch Size' 设置并一次性将所有结果行加载到内存中。当执行获取大量数据集的查询时，这可能导致大量内存使用。有关 PostgreSQL 驱动此行为的更多详细信息，请参阅 https://jdbc.postgresql.org//documentation/head/query.html。支持表达式语言：是（仅使用环境/系统变量求值）
Max Rows Per FlowFile	`0`		单个 FlowFile 中将包含的最大结果行数。这允许您将非常大的结果集拆分为多个 FlowFile。如果指定的值为零，则所有行在单个 FlowFile 中返回。支持表达式语言：是（仅使用环境/系统变量求值）
Output Batch Size	`0`		在提交处理会话之前排队的输出 FlowFile 数量。设置为零时，将在所有结果集行处理完毕且输出 FlowFile 准备好传输到下游关系时提交会话。对于大型结果集，这可能导致在处理器执行结束时大量 FlowFile 突然传输。如果设置了此属性，则当指定数量的 FlowFile 准备好传输时，将提交会话，从而将 FlowFile 释放到下游关系。注意：设置此属性时，不会在 FlowFile 上设置 maxvalue.* 和 fragment.count 属性。支持表达式语言：是（仅使用环境/系统变量求值）
Maximum Number of Fragments	`0`		最大片段数。如果指定的值为零，则返回所有片段。这可以防止此处理器摄取巨大表时出现 OutOfMemoryError。注意：设置此属性可能导致数据丢失，因为传入的结果未排序，并且片段可能在任意边界处结束，其中行不包含在结果集中。支持表达式语言：是（仅使用环境/系统变量求值）
Normalize Table/Column Names	`false`	true false	创建输出模式时是否更改列名称中的字符。例如，冒号和句点将更改为下划线。
Use Avro Logical Types	`false`	true false	是否为 DECIMAL/NUMBER、DATE、TIME 和 TIMESTAMP 列使用 Avro 逻辑类型。如果禁用，则写为字符串。如果启用，则使用逻辑类型并写为其底层类型，具体来说，DECIMAL/NUMBER 写为逻辑 'decimal'：写为字节并带有附加的精度和小数位数元数据，DATE 写为逻辑 'date-millis'：写为自 Unix 纪元（1970-01-01）以来的天数（int），TIME 写为逻辑 'time-millis'：写为自 Unix 纪元以来的毫秒数（int），TIMESTAMP 写为逻辑 'timestamp-millis'：写为自 Unix 纪元以来的毫秒数（long）。如果写入的 Avro 记录的读取器也了解这些逻辑类型，则可以根据读取器实现以更丰富的上下文反序列化这些值。
Default Decimal Precision	`10`		当 DECIMAL/NUMBER 值作为 'decimal' Avro 逻辑类型写入时，需要特定的 'precision'（表示可用数字的位数）。通常，精度由列数据类型定义或数据库引擎默认值决定。但是，某些数据库引擎可能返回未定义的精度（0）。'默认小数精度' 用于写入这些未定义精度的数字。支持表达式语言：是（仅使用环境/系统变量求值）
Default Decimal Scale	`0`		当 DECIMAL/NUMBER 值作为 'decimal' Avro 逻辑类型写入时，需要特定的 'scale'（表示可用小数位数）。通常，小数位数由列数据类型定义或数据库引擎默认值决定。但是，当返回未定义的精度（0）时，某些数据库引擎的小数位数也可能不确定。'默认小数位数' 用于写入这些未定义的数字。如果值的小数位超过指定的小数位数，则值将被四舍五入，例如，小数位数为 0 时 1.53 变为 2，小数位数为 1 时 1.5 保持不变。支持表达式语言：是（仅使用环境/系统变量求值）

动态属性

动态属性允许用户自定义属性的名称与值。

名称	值	描述
initial.maxvalue.<max_value_column>	指定列的初始最大值	指定最大值列的初始最大值。属性应以 `initial.maxvalue.<max_value_column>` 的格式添加。此值仅在首次访问表时使用（当指定了最大值列时）。

关系 (Relationships)

名称	描述	自动终止
`success`	已成功从 SQL 查询结果集创建 FlowFile。	否

写入的 FlowFile 属性

名称	描述
`tablename`	被查询的表名
`querydbtable.row.count`	查询选择的行数
`fragment.identifier`	如果设置了'每个 FlowFile 的最大行数'，则来自同一查询结果集的所有 FlowFile 的 fragment.identifier 属性将具有相同的值。然后可以使用此值来关联结果。
`fragment.count`	这是单个结果集产生的 FlowFile 总数。可以与 fragment.identifier 属性结合使用，以了解有多少 FlowFile 属于同一传入结果集。
`fragment.index`	如果设置了'每个 FlowFile 的最大行数'，则此 FlowFile 在从同一结果集 FlowFile 派生的传出 FlowFile 列表中的位置。可以与 fragment.identifier 属性结合使用，以了解哪些 FlowFile 来自同一查询结果集以及 FlowFile 的产生顺序
`maxvalue.*`	每个属性包含指定'最大值列'的观测最大值。属性的后缀是列名。如果设置了输出批大小，则不会填充此属性。
`mime.type`	将 mime.type 属性设置为Record Writer指定的 MIME 类型。
`record.count`	Record Writer输出的记录数。

状态管理

作用域	描述
CLUSTER	对指定表执行查询后，将保留指定列的最大值以供将来执行查询时使用。这允许处理器仅获取最大值大于保留值的记录。这可用于增量获取、获取新添加的行等。要清除最大值，请按照状态管理文档清除处理器的状态