如何在NiFi中获取ExecuteSQL的COUNT(*)结果并用于PutSQL处理器？

不能直接通过FlowFile属性获取ExecuteSQL执行SELECT COUNT(*)的查询结果——ExecuteSQL处理器会将查询结果以结构化文件内容（如Avro、JSON、CSV等格式）的形式输出到FlowFile中，而非写入FlowFile属性。

要实现将T1的行数存入T2，可按以下步骤操作：

• 步骤1：提取COUNT值到FlowFile属性
  可选两种方式：

  1. 使用ConvertRecord处理器：配置对应ExecuteSQL输出格式的记录读取器，通过UpdateRecord将查询结果中的COUNT(*)字段值提取为FlowFile属性（比如命名为t1_row_count）。
  2. 使用ExecuteScript处理器：编写简单脚本读取FlowFile内容并提取COUNT值到属性，示例Groovy脚本如下：

     import groovy.json.JsonSlurper
     
     def flowFile = session.get()
     if (!flowFile) return
     
     def slurper = new JsonSlurper()
     def content = flowFile.read().withReader("UTF-8") { slurper.parse(it) }
     def count = content[0]["COUNT(*)"]
     
     flowFile = session.putAttribute(flowFile, "t1_row_count", count.toString())
     session.transfer(flowFile, REL_SUCCESS)
     

• 步骤2：用PutSQL写入T2
  在PutSQL处理器中开启「支持表达式语言」选项，编写SQL语句引用已提取的属性：

  INSERT INTO T2 (row_count, create_time) VALUES (${t1_row_count}, CURRENT_TIMESTAMP)
  

内容的提问来源于stack exchange，提问作者Eren Irmak