Terraform 在 Kinesis Firehose 动态分区上抛出 InvalidArgumentException Duplicate ProcessorParameter passed to ProcessingConfiguration

Question

我正在尝试使用 terraform 创建一个 Kinesis Firehose，并使用我收到的 JSON 中的两个分区查询进行动态分区，我的处理配置如下所示

processing_configuration {
  enabled = true
  processors {
    type = "RecordDeAggregation"
    parameters {
      parameter_name  = "SubRecordType"
      parameter_value = "JSON"
    }
  }
  processors {
    type = "MetadataExtraction"
    parameters {
      parameter_name  = "JsonParsingEngine"
      parameter_value = "JQ-1.6"
    }
    parameters {
      parameter_name  = "MetadataExtractionQuery"
      parameter_value = "{transaction_id:.transaction_id}"
    }
    parameters {
      parameter_name  = "MetadataExtractionQuery"
      parameter_value = "{stage:.stage}"
    }
  }
}

但是当我执行这部分代码时，它会返回处理配置的重复错误。

我还尝试为新的 ExtractionQuery 创建一个应用程序处理器，它看起来像这样

processing_configuration {
  enabled = true
  processors {
    type = "RecordDeAggregation"
    parameters {
      parameter_name  = "SubRecordType"
      parameter_value = "JSON"
    }
  }
  processors {
    type = "MetadataExtraction"
    parameters {
      parameter_name  = "JsonParsingEngine"
      parameter_value = "JQ-1.6"
    }
    parameters {
      parameter_name  = "MetadataExtractionQuery"
      parameter_value = "{transaction_id:.transaction_id}"
    }
  }
  processors {
    type = "MetadataExtraction"
    parameters {
      parameter_name  = "JsonParsingEngine"
      parameter_value = "JQ-1.6"
    }
    parameters {
      parameter_name  = "MetadataExtractionQuery"
      parameter_value = "{stage:.stage}"
    }
  }
}

但它失败并出现错误，指出只允许使用一个 MetadataExtraction 处理器。

Answer 1

通过使用 JQ 格式将两个查询合并为一个来解决，这样 firehose 就会将它们分开，使用此代码段进行尝试并成功。

processing_configuration {
  enabled = true
  processors {
    type = "RecordDeAggregation"
    parameters {
      parameter_name  = "SubRecordType"
      parameter_value = "JSON"
    }
  }
  processors {
    type = "MetadataExtraction"
    parameters {
      parameter_name  = "JsonParsingEngine"
      parameter_value = "JQ-1.6"
    }
    parameters {
      parameter_name  = "MetadataExtractionQuery"
      parameter_value = "{transaction_id:.transaction_id,stage:.stage}"
    }
  }
}