使用 Apache Nifi 进行具有关系的 ETL 多个表

Question

我想使用 Apache Nifi 将数据从 Oracle 数据库 ETL 到 MS SQL Server 数据库。

目标数据库，有 2 个表（有关系）和一个 1-n 基数（n 值很大，比如 20k 之类的），我们将表命名为

Header

和

Detail

。

我在源 (Oracle) 数据库上创建了 2 个视图，一个返回

Header

数据，一个返回

Detail

数据。

因此，我需要从 Oracle 视图中获取所有

Header

，将每条记录插入到 SQL Server 中，然后对于每条记录，从 Oracle 中获取所有

Detail

并将它们插入到 SQL Server 中。

NiFi 中迭代记录的最佳解决方案是什么？

由于我在标头中有很多记录，在详细信息中有很多记录（正如我所说，标头的每条记录在详细信息中可能有 20k 条记录），加入所有记录并获取所有记录根本效率不高。

这是应该定期运行的作业。

因此，我开始使用

QueryDatabaseTable

处理器来读取

Header

视图（并且它会增量工作），并使用

PutDatabaseRecord

处理器填充 MS SQL Server 中的

Header

表。效果很好。

现在我想通过

Detail

处理器获取

ExecuteSQL

视图的记录，问题是现在我需要拥有

ID

的每个记录的

Header

将其传递给

ExecuteSQL

的查询，我有写了类似

select * from Detail d where d.header_id = ${ID}

的东西，但失败了，因为

${id}

是空的。

我在将

SplitAvro

插入目标数据库后添加了一个

Header

进程，好处是现在我有一堆FlowFiles而不是一个大的（每条记录1个），然后我添加了一个

FilterAttribute

进程，所以我可以只拥有

ID

属性。

现在我不知道如何将该属性传递给下一个

ExecuteSQL

进程....

Answer 1

我已经完成了工作，步骤如下：

1. 执行 SQL：仅获取非 ETLed 标头 ID，
2. 转换记录：从 Avro 到 JSON
3. 拆分 JSON
4. 提取文本：将标头 ID 提取到属性，因为属性可以传递给查询
5. 执行SQL：获取头记录（一条一条）
6. Put Database Record：插入表头记录
7. 执行SQL：获取表头所有明细记录
8. 放入数据库记录：批量插入明细记录
9. 放置SQL：标记头记录，它是ETLed的，所以在以后的执行中不会被选中。