NiFi ValidateRecord校验CSV转Avro因字段含双引号路由失败求助

我之前处理过类似的CSV解析场景，NiFi的CSVReader是严格遵循RFC 4180标准的——按照标准，字段内的双引号必须用另一个双引号转义（比如把字段里的"Accept Response"改成""Accept Response""），否则会被判定为格式错误，导致FlowFile路由到Failure关系。既然你不想篡改原始字段值，这里有几个可行的解决思路：

1. 调整CSVReader控制器服务的配置（针对标准转义场景） #

先检查你的CSVReader的Escape Character属性，默认可能是空值。如果把它设置为双引号（"），CSVReader就会识别字段内的双引号转义（即两个连续双引号代表一个实际双引号）。不过这个方法只适用于符合标准转义规则的CSV，如果你的数据里是单个未转义的双引号，这个配置还是会解析失败，因为它不符合标准语法。

2. 使用ScriptedReader自定义解析逻辑（灵活处理非标准CSV） #

如果你的CSV是“非标准”的（字段内有单个未转义双引号），可以用ScriptedReader控制器服务，写Groovy或Python脚本自定义解析逻辑，绕过严格的标准校验。

举个Groovy脚本的简单示例，核心是用正则匹配正确的字段边界，避免把字段内的双引号当成字段结束符：

import org.apache.nifi.serialization.record.*
import java.util.regex.*

def reader = new BufferedReader(new InputStreamReader(inputStream, charset))
def headerLine = reader.readLine()
def headers = headerLine.split(',')*.replaceAll('^"|"$', '') // 去除首尾引号

def recordSchema = new SimpleRecordSchema(headers.collect { new RecordField(it, RecordFieldType.STRING.getDataType()) })
def recordList = []

def line
while ((line = reader.readLine()) != null) {
    def pattern = Pattern.compile('(?<=^|,)(?:"([^"]*(?:"[^"]*)*)"|([^,]*))(?=,|$)')
    def matcher = pattern.matcher(line)
    def fields = []
    while (matcher.find()) {
        def fieldValue = matcher.group(1) ?: matcher.group(2)
        fields.add(fieldValue?.replaceAll('""', '"')) // 处理转义的双引号
    }
    def record = new MapRecord(recordSchema, headers.zip(fields).collectEntries())
    recordList.add(record)
}

return new SimpleRecordSet(recordSchema, recordList.iterator())

这个脚本会正确解析字段内包含单个双引号的CSV行，同时保留原始的字段值。

3. 临时替换+还原（不篡改最终数据的折中方案） #

如果不想写自定义脚本，可以用ReplaceText做临时转义，解析完成后再还原原始值：

• 第一步：临时转义：用ReplaceText处理器，配置Search Value为(?<=[^",])"(?=[^",])（匹配字段内的单个双引号，排除字段首尾的引号），Replacement Value为""（符合RFC标准的转义格式）。这一步只是为了让CSVReader能正确解析，不会修改原始FlowFile（可以用ReplaceText的Replacement Strategy为Replace All Occurrences）。
• 第二步：校验+转换：用ValidateRecord处理器完成校验和CSV转Avro。
• 第三步：还原原始值：用ReplaceTextRecord处理器，针对有问题的字段，把Avro里的""替换回"，这样最终的Avro数据就和原始CSV的字段值完全一致了。

这个方法的好处是不需要写复杂脚本，只需要配置几个处理器，就能在不破坏原始数据的前提下完成解析校验。

内容的提问来源于stack exchange，提问作者Vikramsinh Shinde