删除nifi中的重复项

8iwquhpp  于 2021-06-20  发布在  Mysql
关注(0)|答案(1)|浏览(333)

我有一个流querydatabasetable->convertrecord->putelasticseachhttprecord我想做的是从mysql数据库中获取完整的数据,并将其输入elasticsearch,使用kibana对其进行分析。但是,我的数据有如下重复的列:(以黑色突出显示的是唯一的重复值)
id,机器名称,值1,值2,日期
1,,10,34,2018-09-27 10:40:10
2,,10,34,2018-09-27 10:41:14
3,,10,34,2018-09-27 10:42:19
2018年9月27日10:45:19
所以在我的表中,id是主键,timestamp字段不断更新。我想要实现的是只获取特定机器名的一条记录。下面的示例显示了我想要的输出表:
id,机器名称,值1,值2,日期
1,,10,34,2018-09-27 10:40:10
2018年9月27日10:45:19
如何在nifi实现这一点?目的是删除重复的列。如果可能的话,请告诉我使用哪个处理器和设置什么配置?
queryrecord处理器中出现以下错误:



任何建议都非常感谢。谢谢您

dwthyt8l

dwthyt8l1#

而不是使用 convertrecord 处理器,使用queryrecord处理器。
使用row\ u number()窗口函数和 Partition by Machine Name,value1,value2...etc 并只选择每个分区数据中的第一行。
查询记录配置:
我试过了 Csv reader 以及 Json set Writer 控制器服务
查询:

select id,machinename,value1,value2,"date" from(
SELECT id,
         machinename,value1,value2,"date",
         row_number() over (partition by machinename order by "date" asc) as rn
  from FLOWFILE
) sq
WHERE rn = 1


输出:

[ {
  "id" : "1",
  "machinename" : "abc",
  "value1" : "10",
  "value2" : "34",
  "date" : "2018-09-27 10:40:10"
}, {
  "id" : "4",
  "machinename" : "xyz",
  "value1" : "12",
  "value2" : "45",
  "date" : "2018-09-27 10:45:19"
} ]

流量:

QueryDatabaseTable->QueryRecord->PutElasticseachHttpRecord

如果你有空间 machine name column name 那么 enclose column name in double quotes("<col_name>") 使用下面的查询

select id,"machine name",value1,value2,"date" from(
SELECT id,
         "machine name",value1,value2,"date",
         row_number() over (partition by "machine name" order by "date" asc) as rn
  from FLOWFILE
) sq
WHERE rn = 1

我只包括 machine name 列,但可以包括其他列 value1,value2 根据您的要求。
请使用此模板以获取更多参考。

相关问题