求助：在StreamSets中将动态字段转为列名、值字符串

步骤1：添加Groovy脚本处理器到流水线 #

因为你的字段是动态可变的，字段映射器这类静态处理器无法满足需求，需要用Groovy脚本处理器来遍历所有字段并动态生成目标串。

步骤2：使用以下Groovy脚本 #

直接复制粘贴到脚本处理器的编辑框中，无需修改（若有特殊字段过滤需求可参考后续说明）：

// 初始化存储列名和值的列表
def columnNames = []
def values = []

// 遍历当前记录的所有字段
record.fields.each { key, value ->
    // 字段名包裹单引号后加入列表
    columnNames.add("'${key}'")
    // 处理字段值：null转为SQL的NULL，非null值包裹单引号并转义内部单引号
    def valStr = value == null ? "NULL" : "'${value.toString().replace("'", "''")}'"
    values.add(valStr)
}

// 将列表转为逗号分隔的字符串，生成目标字段
record.value['ColumnName'] = columnNames.join(', ')
record.value['ValueName'] = values.join(', ')

// 返回处理后的记录
return [record]

步骤3：验证输出 #

运行流水线后，每条记录会新增两个字段：

• ColumnName：格式为 'Source_id', 'Partition_id', 'Key'
• ValueName：格式为 '1234', 'ABC', 'W3E'

额外说明 #

• 若需要排除某些固定字段，可在遍历逻辑中添加过滤，比如排除名为timestamp的字段：

  record.fields.each { key, value ->
      if (!key.equalsIgnoreCase("timestamp")) {
          // 原有的添加列名和值的逻辑
      }
  }
  

• 脚本自动处理了值中的单引号转义（转为双单引号），避免SQL语法错误或注入风险；同时处理了null值，转为SQL兼容的NULL。

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