如何在 Jython 中使用修改后的数据更新行？

我有一个 csv 文件，其中包含数十万行，下面是一些示例行..，

1,Ni,23,28-02-2015 12:22:33.2212-02
2,Fi,21,28-02-2015 12:22:34.3212-02
3,Us,33,30-03-2015 12:23:35-01
4,Uk,34,31-03-2015 12:24:36.332211-02

我需要获取日期时间格式错误的 csv 数据的最后一列。所以我需要获得默认的日期时间格式（"YYYY-MM-DD hh:mm:ss[.nnn]"）来自数据的最后一列。

我尝试了以下脚本来从中获取行并写入流文件。

import json
import java.io
from org.apache.commons.io import IOUtils
from java.nio.charset import StandardCharsets
from org.apache.nifi.processor.io import StreamCallback

class PyStreamCallback(StreamCallback):
  def __init__(self):
        pass
  def process(self, inputStream, outputStream):
    text = IOUtils.readLines(inputStream, StandardCharsets.UTF_8)
    for line in text[1:]:
        outputStream.write(line + "\n") 

flowFile = session.get()
if (flowFile != None):
  flowFile = session.write(flowFile,PyStreamCallback())
  flowFile = session.putAttribute(flowFile, "filename", flowFile.getAttribute('filename'))
  session.transfer(flowFile, REL_SUCCESS)

但我无法找到一种方法将其转换为如下输出。

1,Ni,23,28-02-2015 12:22:33.221
2,Fi,21,29-02-2015 12:22:34.321
3,Us,33,30-03-2015 12:23:35
4,Uk,34,31-03-2015 12:24:36.332

我已经和我的朋友（谷歌）检查了解决方案，但仍然找不到解决方案。

谁能指导我将这些输入数据转换为我所需的输出？

在这个转换中，不必要的数据位于每行的末尾，因此使用正则表达式管理转换任务非常容易。

^(.*:\d\d)((\.\d{1,3})(\d*))?(-\d\d)?

在这里检查正则表达式和解释：https://regex101.com/r/sAB4SA/2

一旦您有一个大文件 - 最好不要将其加载到内存中。以下代码将整个文件加载到内存中：

IOUtils.readLines(inputStream, StandardCharsets.UTF_8)

最好逐行迭代。

所以这段代码是为了ExecuteScript使用 python (Jython) 语言的 nifi 处理器：

import sys
import re
import traceback
from org.apache.commons.io import IOUtils
from org.apache.nifi.processor.io import StreamCallback
from org.python.core.util import StringUtil
from java.lang import Class
from java.io import BufferedReader
from java.io import InputStreamReader
from java.io import OutputStreamWriter


class TransformCallback(StreamCallback):
    def __init__(self):
        pass

    def process(self, inputStream, outputStream):
        try:
            writer = OutputStreamWriter(outputStream,"UTF-8")
            reader = BufferedReader(InputStreamReader(inputStream,"UTF-8"))
            line = reader.readLine()
            p = re.compile('^(.*:\d\d)((\.\d{1,3})(\d*))?(-\d\d)?')
            while line!= None:
                # print line
                match = p.search(line)
                writer.write( match.group(1) + (match.group(3) if match.group(3)!=None else '') )
                writer.write('\n')
                line = reader.readLine()
            writer.flush()
            writer.close()
            reader.close()
        except:
            traceback.print_exc(file=sys.stdout)
            raise


flowFile = session.get()
if flowFile != None:
    flowFile = session.write(flowFile, TransformCallback())

    # Finish by transferring the FlowFile to an output relationship
    session.transfer(flowFile, REL_SUCCESS)

一旦问题涉及到 nifi，这里有一些似乎更简单的替代方案

与上面相同的代码，但在 groovy for nifi 中ExecuteScript处理器：

def ff = session.get()
if(!ff)return
ff = session.write(ff, {rawIn, rawOut->
    // ## transform streams into reader and writer
    rawIn.withReader("UTF-8"){reader->
        rawOut.withWriter("UTF-8"){writer->
            reader.eachLine{line, lineNum->
                if(lineNum>1) { // # skip the first line
                    // ## let use regular expression to transform each line
                    writer << line.replaceAll( /^(.*:\d\d)((\.\d{1,3})(\d*))?(-\d\d)?/ , '$1$3' ) << '\n'
                }
            }
        }
    }
} as StreamCallback)
session.transfer(ff, REL_SUCCESS)

替换文本处理器

如果正则表达式没问题 - nifi 中最简单的方法是ReplaceText可以进行正则表达式逐行替换的处理器。

在这种情况下，您不需要编写任何代码，只需构建正则表达式并正确配置您的处理器即可。