使用StreamSets Groovy Evaluator修改XML值遇路径不可达错误求助

在使用StreamSets处理XML数据时，尝试通过Groovy Evaluator修改XML中/SalesData/Transaction/Item/Price/Data字段的值，但报错提示字段路径不可达。流水线包含FTP源、Groovy Evaluator、Trash三个阶段，因后续需复杂转换必须使用Groovy Evaluator，已确认XML结构与字段路径一致，但仍报错。

XML记录结构 #

<SalesData>
  <Transaction>
    <Item>
      <Price>
        <Data>123</Data>
      </Price>
    </Item>
  </Transaction>
  <Transaction>
    <Item>
      <Price>
        <Data>456</Data>
      </Price>
    </Item>
  </Transaction>
</SalesData>

使用的Groovy代码 #

import com.streamsets.pipeline.api.Field

records = sdc.records
for (record in records) {
    try {
      record.sdcRecord.set('/SalesData/Transaction/Item/Price/Data', '000')
            
      output.write(record)

    } catch (e) {
        log.error("Error processing record: ${e.toString()}", e)
        error.write(record, e.toString())
    }
}

报错信息 #

Record1-Error Record1 SCRIPTING_04 - Script sent record to error: java.lang.IllegalArgumentException: Field-path '/SalesData/Transaction/Item/Price/Data' not reachable

核心原因 #

FTP源默认会将整个XML作为单一字符串字段（通常是/text字段）传入下游，而非解析成结构化的层级字段。直接用结构化字段路径访问自然会提示路径不可达。

解决步骤 #

1. 添加XML解析器

   • 在FTP源和Groovy Evaluator之间插入XML Parser处理器。
   • 配置XML Parser：
     • 设置Field to Parse为FTP源输出的XML字符串字段（通常是/text）。
     • 设置Target Field为根节点路径（比如/SalesData），或留空让解析后的结构直接替换原字段。
     • 根据需求配置Parse Multiple Records：若每个XML文件包含多个Transaction且需拆分，开启该选项并指定Record XPath为/SalesData/Transaction；若需保留完整结构则关闭。

2. 修改Groovy代码适配结构

   • 若保留完整XML结构（未拆分Transaction），因Transaction是数组类型，需遍历修改每个元素：

     import com.streamsets.pipeline.api.Field
     
     records = sdc.records
     for (record in records) {
         try {
             def transactions = record.sdcRecord.get('/SalesData/Transaction')
             if (transactions != null && transactions.type == Field.Type.LIST) {
                 transactions.value.each { transaction ->
                     transaction.set('/Item/Price/Data', Field.create('000'))
                 }
             }
             output.write(record)
         } catch (e) {
             log.error("Error processing record: ${e.toString()}", e)
             error.write(record, e.toString())
         }
     }
     

   • 若已通过XML Parser拆分出单个Transaction记录，代码可简化为：

     import com.streamsets.pipeline.api.Field
     
     records = sdc.records
     for (record in records) {
         try {
             record.sdcRecord.set('/Item/Price/Data', '000')
             output.write(record)
         } catch (e) {
             log.error("Error processing record: ${e.toString()}", e)
             error.write(record, e.toString())
         }
     }
     

3. 验证字段路径

   • 在Groovy Evaluator前添加Preview阶段，查看解析后的字段结构，确认路径是否正确。比如拆分后每个记录的根是Transaction，路径就变为/Item/Price/Data而非完整层级。

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