Terraform 抛出 InvalidArgumentException 传递到 Kinesis Firehose 动态分区上的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 创建一个 appart 处理器，它看起来像这样

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}"
    }
  }
}