pentaho - pentaho 数据集成中新插入或更新的行数

Question

我是 Pentaho 数据集成的新手；我需要将一个数据库作为 ETL 作业集成到另一个位置。我想计算 ETL 作业期间插入/更新的次数，并将该计数插入到另一个表中。谁能帮我解决这个问题吗？

Answer 1

我认为到目前为止，PDI 中没有内置功能可以返回插入/更新步骤的受影响的行数。

尽管如此，大多数数据库供应商都能够为您提供从给定操作获取受影响行数的能力。

例如，在 PostgreSQL 中，它看起来像这样:

/* Count affected rows from INSERT */
WITH inserted_rows AS (
    INSERT INTO ...
    VALUES
        ...
    RETURNING 1
)
SELECT count(*) FROM inserted_rows;

/* Count affected rows from UPDATE */
WITH updated_rows AS (
    UPDATE ...
    SET ...
    WHERE ...
    RETURNING 1
)
SELECT count(*) FROM updated_rows;

但是，您的目标是在 PDI 作业中执行此操作，因此我建议您尝试达到控制 SQL 脚本的程度。

建议:将源数据保存在目标数据库服务器上的文件中，然后使用它(也许具有批量加载功能)来插入/更新，然后将受影响的行数保存到PDI 变量。请注意，您可能需要在作业范围内使用 SQL 脚本步骤。

编辑:实现是选择设计的问题，因此建议的解决方案是众多解决方案之一。在非常高的层面上，您可以执行如下操作。

• 转换 I - 从源中提取数据
  • 从源获取数据，无论是数据库还是其他任何数据
  • 以适合目标数据库结构的方式准备输出
  • 使用文件系统上的文本文件输出步骤保存 CSV 文件
• 家长工作
  • 如果 PDI 服务器与目标 DB 服务器相同:
    • 使用“执行 SQL 脚本”步骤可以:
      • 从文件中读取数据并执行插入/更新
      • 将受影响的行数写入表中(理想情况下，该表还可以包含操作的时间戳，以便您可以跟踪情况)
  • 如果 PDI 服务器与目标数据库服务器不同:
    • 将源数据文件上传到服务器，例如FTP/SFTP 文件上传步骤
    • 使用“执行 SQL 脚本”步骤可以:
      • 从文件中读取数据并执行插入/更新
      • 将受影响的行数写入表格

编辑2:另一个建议的解决方案

根据 @user3123116 的建议，您可以使用“比较字段”步骤(如果不是您环境的一部分，请检查市场)。

我看到的唯一缺点是您必须在插入/更新之前查询目标数据库，这当然性能较差。

最终它可能看起来像这样(请注意，这只是比较和计数部分):

另请注意，您可以拆分源数据流的输入(复制，而不是分发)，并执行插入/更新，但该流必须等待字段比较的流结束对目标数据库的查询，否则可能会得到错误的统计结果。