NiFi GenerateTableFetch 不存储每个 database.name 的状态 [英] NiFi GenerateTableFetch does not store state per database.name

问题描述

我正在测试 NiFi 以替换我们当前的摄取设置，该设置从表的多个 MySQL 分片导入数据并将其存储在 HDFS 中.

我正在使用 GenerateTableFetch 和 ExecuteSQL 来实现这一点.

每个传入的流文件都有一个 database.name 属性，DBCPConnectionPoolLookup 使用该属性来选择相关的分片.

问题是，假设我有 2 个分片可以从中提取数据，shard_1 和 shard_2 用于表帐户，而且我还有 updated_at作为 Maximum Value Columns，它不会为每个分片的 table@updated_at 存储状态.每个表只有 1 个条目处于状态.

当我签入 Data Provenance 时，我看到 shard_2 流文件文件被丢弃而没有传递给 ExecuteSQL.我的猜测是因为 shard_1 查询首先被执行，然后当 shard_2 查询到来时，它的记录会根据 shard_1 的 updated_at 进行检查，因为它返回空，所以它会删除文件.

有人遇到过这个问题吗?还是我遗漏了什么?

解决方案

在数据库获取处理器(例如 QueryDatabaseTable、GenerateTableFetch)中存储状态的方案之后，添加了通过 DBCPConnectionPoolLookup 选择不同数据库的能力.此外，获取数据库名称因 RDBMS 驱动程序而异，它可能在 DatabaseMetaData 或 ResultSetMetaData 中，可能在 getCatalog() 或 getSchema() 中，也可能不在.

我已经编写了 NIFI-5590 来介绍这一改进.>

I am testing out NiFi to replace our current ingestion setup which imports data from multiple MySQL shards of a table and store it in HDFS.

I am using GenerateTableFetch and ExecuteSQL to achieve this.

Each incoming flow file will have a database.name attribute which is being used by DBCPConnectionPoolLookup to select the relevant shard.

Issue is that, let's say I have 2 shards to pull data from, shard_1 and shard_2 for table accounts and also I have updated_at as Maximum Value Columns, it is not storing state for the for the table@updated_at per shard. There is only 1 entry per table in state.

When I check in Data Provenance, I see the shard_2 flowfile file getting dropped without being passed to ExecuteSQL. And my guess is it's because shard_1 query gets executed first and then when shard_2 query comes, it's records are checked against shard_1's updated_at and since it returns empty, it drops the file.

Has anyone faced this issue? Or am I missing something?

解决方案

The ability to choose different databases via DBCPConnectionPoolLookup was added after the scheme to store state in the database fetch processors (QueryDatabaseTable, GenerateTableFetch, e.g.). Also, getting the database name differs between RDBMS drivers, it might be in the DatabaseMetaData or ResultSetMetaData, possibly in getCatalog() or getSchema() or neither.

I have written NIFI-5590 to cover this improvement.

这篇关于NiFi GenerateTableFetch 不存储每个 database.name 的状态的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！