python之Apache NiFi : Processing multiple csv's using the ExecuteScript Processor

我有一个包含 70 列的 csv。第 60 列包含一个值，它决定记录是 valid 还是 invalid。如果第 60 列有 0、1、6 或 7，则它是 有效的。如果它包含任何其他值，则它的无效。

我意识到完全依赖于更改 Apache NiFi 中处理器的属性是不可能实现此功能的。因此，我决定使用 executeScript 处理器 并将这段 python 代码添加为文本正文。

import csv 
 
valid =0 
invalid =0 
total =0 
file2 = open("invalid.csv","w") 
file1 = open("valid.csv","w") 
 
with  open('/Users/himsaragallage/Desktop/redder/Regexo_2019101812750.dat.csv') as f: 
    r = csv.reader(f) 
    for row in f: 
        # print row[1] 
        total +=1 
 
        if row[59] == "0" or row[59] == "1" or row[59] == "6" or row[59] == "7": 
            valid +=1 
            file1.write(row) 
        else: 
            invalid += 1 
            file2.write(row) 
file1.close() 
file2.close() 
print("Total : " + str(total)) 
print("Valid : " + str(valid)) 
print("Invalid : " + str(invalid)) 

我不知道如何在 executeScript 处理器中使用 session 和代码，如 this question 所示。 .所以我只是写了一个简单的python代码并将有效和无效数据定向到不同的文件。我使用的这种方法有许多限制。

1. 我希望能够动态处理具有不同文件名的 csv。
2. 将无效数据发送到的 csv 文件名也必须与输入 csv 文件名相同。
3. 我的 redder 文件夹中大约有 20 个 csv。所有这些都必须一次性处理。

希望您能为我建议一种方法来执行以下操作。随时通过编辑我使用的 python 代码或什至完全使用一组不同的处理器并完全排除使用 ExecuteScript Processer

请您参考如下方法:

这是有关 how to use QueryRecord processor 的完整分步说明

基本上，您需要设置突出显示的属性