/ /

`$https` Stage de agregación (Stream Processing)

Definición

$https

La etapa $https especifica una conexión en el Registro de conexiones para enviar solicitudes HTTPS. Cada vez que una etapa anterior pasa un documento a $https, la etapa envía una nueva solicitud.

Ubicación

$https debe venir después de la etapa $source, puede venir antes de la etapa $emit o $merge, o puede servir como la etapa final en una pipeline. Puedes utilizar $https en los pipelines internos de Stream Processor Windows.

Sintaxis

Para enviar una solicitud HTTPS a una conexión dada:

{
  "$https": {
    "connectionName": "<registered-connection>",
    "path" : "<subpath>" | <expression>,
    "parameters" : {
      "<key1>" : "<val1>",
      . . .
      "<keyn>" : "<valn>"
    },
    "method" : "<GET | POST | PUT | PATCH | DELETE>",
    "headers" : {
      "<key1>" : "<val1>",
      . . .
      "<keyn>" : "<valn>"
    },
    "as" : "response",
    "onError" : "<DLQ | Ignore | Fail>",
    "payload" : [{
      <inner-pipeline>
    }],
    "config" : {
      "connectionTimeoutSec" : <integer>,
      "requestTimeoutSec" : <integer>,
      "parseJsonStrings": <boolean>
    }
  }
}

La etapa $https procesa un documento con los siguientes campos:

Campo	Tipo	Necesidad	Descripción
`connectionName`	string	Requerido	Etiqueta que identifica la conexión en el Registro de conexiones a la que se enviará una solicitud de `HTTPS`.
`path`	string \| expresión	Opcional	Ruta para añadir a la URL a la que apunta tu `connectionName`. Por ejemplo, si especificas un `connectionName` que se resuelve a `https://sample.com`, puedes especificar un `path` de `"endpoint"` para que el procesador de flujo envíe `HTTPS` solicitudes a `https://sample.com/endpoint`. Si defines `path` como una expresión, dicha expresión debe evaluarse como una string. El endpoint de la API al que llamas debe ser idempotente.
`parameters`	Documento	Opcional	Documento que contiene pares clave-valor para pasar como parámetros a tu llamada al endpoint de API. Cada clave debe ser un string, y cada valor debe evaluarse como un valor numérico, de string o booleano. Este campo admite expresiones como valores.
`method`	string	Opcional	Método de solicitud HTTPS para su conexión. Debe ser uno de los siguientes valores: `"GET"` `"POST"` `"PUT"` `"PATCH"` `"DELETE"` Se establece por defecto en `"GET"`.
`headers`	Documento	Opcional	Documento que contiene pares clave-valor para pasar como encabezados al endpoint de la API. Cada clave debe ser una cadena y cada valor debe evaluar a una cadena. Este campo admite expresiones como valores. Si el endpoint de la API requiere autenticación, como una clave de API o un token de acceso Bearer, es necesario agregar los detalles de autenticación como encabezados al definir la conexión, para evitar proporcionarlos como texto plano dentro de este operador. Los nombres y valores de encabezados HTTP no válidos no se envían al punto de conexión de la API. En cambio, estos son ignorados. Para aprender más sobre HTTP headers inválidos, se puede consultar RFC 9110. Si una expresión en un valor falla o se evalúa a un tipo diferente de una string, el mensaje se envía a la fila de letra muerta y el operador no envía esta solicitud al endpoint de la API.
`as`	string	Condicional	Nombre del campo para la respuesta de la REST API. El parámetro `as` es obligatorio cuando se utiliza como etapa no sumidero (intermedia) y opcional cuando se utiliza como última etapa o etapa de sumidero. Si el endpoint devuelve 0 bytes, el operador no configura el campo `as`. El operador admite respuestas con un `Content-Type` de `application/json` o `text/plain`. Si el endpoint de la API devuelve una respuesta con un `Content-Type` diferente, el operador gestiona el documento según el comportamiento de `onError` que definas. Si el endpoint de la API devuelve una respuesta sin un `Content-Type` definido, el operador asume que la respuesta es `application/json`.
`onError`	string	Opcional	Comportamiento cuando el operador encuentra una falla relacionada con `HTTPS`. Debe ser uno de los siguientes valores: `"dlq"` : Pase el documento afectado a la fila de letra muerta. `"ignore"` : Ignora el error y pasa el documento afectado a la siguiente etapa del pipeline. `"fail"` : Finalizar el procesador de flujo en caso de error. El operador considera todos los `2XX` códigos de estado HTTP como éxitos. Si el operador recibe alguno de los siguientes códigos de estado HTTP como respuesta, el operador sigue el comportamiento en función del valor que usted proporciona en este campo: `400` `404` `410` `413` `414` `431` El operador considera cualquier otro código de estado HTTP como un error `"fail"`. Por ejemplo, si el endpoint de la API devuelve un código de estado HTTP `500`, el procesador entra en un estado fallido y se detiene. `onError` no se activa en errores que surjan de una configuración incorrecta del propio operador `$https`, como expresiones no válidas. Se establece por defecto en `"dlq"`.
`payload`	arreglo	Opcional	Pipeline interno personalizado que permite personalizar el cuerpo de la solicitud enviado al endpoint de la API. `payload` soporta las siguientes expresiones: `$project` `$addFields` `$replaceRoot` `$set` Por defecto, se envía el mensaje completo al endpoint de API. Atlas Stream Processing envía una carga útil JSON en modo relajado al endpoint API. Las solicitudes HTTP no válidas no son enviadas al endpoint de la API. En cambio, estos se envían a la fila de letra muerta. Para obtener más información sobre los cuerpos de solicitudes HTTP no válidos, consulte RFC 9110.
`config`	Documento	Opcional	Documento que contiene campos que sobrescriben diversos valores por defecto.
`config.connectionTimeoutSec`	entero	Opcional	El tiempo, en segundos, tras el cual un intento de conexión `HTTPS` se agota si no se puede establecer una conexión. Se establece por defecto en `30`.
`config.parseJsonStrings`	booleano	Opcional	Configuración que determina si Atlas itera recursivamente a través de la respuesta del servidor y serializa los valores de string que contienen JSON válido (utilizando comillas escapadas) en BSON válido para que pueda trabajar con los resultados en las siguientes etapas de la pipeline. Se establece por defecto en `false`.
`config.requestTimeoutSec`	entero	Opcional	El tiempo, en segundos, tras el cual una solicitud `HTTPS` supera el límite si no se recibe respuesta. Se establece por defecto en `60`.

Ejemplos

Una fuente de datos de transmisión genera informes meteorológicos detallados desde varias ubicaciones, cumpliendo con el esquema del Conjunto de Datos Meteorológicos de Muestra. La siguiente agregación tiene tres etapas:

La etapa $source establece una conexión con el Apache Kafka broker que recopila estos informes en un tema denominado my_weatherdata, exponiendo cada registro a medida que se procesa a las siguientes etapas de agregación. Este etapa también reemplaza el nombre del campo de marca de tiempo que proyecta, estableciéndolo en ingestionTime.
Para cada registro del broker de Apache Kafka, la etapa $https envía una solicitud a una fuente meteorológica HTTPS definida en la conexión https_weather. La solicitud utiliza position.coordinates del registro en la solicitud HTTPS para recopilar un pronóstico de temperatura máxima de siete días en grados Celsius para esa ubicación, que se añade al documento del pipeline en un campo airTemperatureForecast.
La fase $merge escribe la salida en una colección Atlas denominada stream en la base de datos sample_weatherstream. Si no existe tal base de datos o colección, Atlas los creará.

{
  '$source': {
    connectionName: 'sample_weatherdata',
    topic: 'my_weatherdata',
    tsFieldName: 'ingestionTime'
  }
},
{
  '$https': {
    connectionName: 'https_weather',
    path: 'forecast',
    parameters: {
      latitude: { $arrayElemAt: ['$$ROOT.position.coordinates', 0 ] },
      longitude: { $arrayElemAt: ['$$ROOT.position.coordinates', 1 ] }
    },
    as: 'airTemperatureForecast'
},
{
  '$merge': {
    into: {
     connectionName: 'weatherStreamOutput',
     db: 'sample_weatherstream',
     coll: 'stream'
    }
  }
}

Para ver los documentos de la colección sample_weatherstream.stream resultante, conéctate a un clúster de Atlas y ejecuta el siguiente comando:

db.getSiblingDB("sample_weatherstream").stream.find()

{
  _id: ObjectId('66ad2edfd4fcac13b1a28ce3'),
  airTemperature: { quality: '1', value: 27.7 },
  airTemperatureForecast: [22.3, 22.4, 22.5, 22.3, 22.4, 22.5, 23.1],
  atmosphericPressureChange: {
    quantity24Hours: { quality: '9', value: 99.9 },
    quantity3Hours: { quality: '1' },
    tendency: { code: '1', quality: '1' }
  },
  atmosphericPressureObservation: {
    altimeterSetting: { quality: '1', value: 1015.9 },
    stationPressure: { quality: '1', value: 1021.9 }
  },
  callLetters: 'CGDS',
  dataSource: '4',
  dewPoint: { quality: '9', value: 25.7 },
  elevation: 9999,
  extremeAirTemperature: { code: 'N', period: 99.9, quantity: '9', value: -30.4 },
  ingestionTime: ISODate('2024-08-02T19:09:18.071Z'),
  liquidPrecipitation: { condition: '9', depth: 160, period: 24, quality: '2' },
  pastWeatherObservationManual: {
    atmosphericCondition: { quality: '1', value: '8' },
    period: { quality: '9', value: 3 }
  },
  position: { coordinates: [ 30.27, -97.74], type: 'Point' },
  precipitationEstimatedObservation: { discrepancy: '4', estimatedWaterDepth: 4 },
  presentWeatherObservationManual: { condition: '53', quality: '1' },
  pressure: { quality: '1', value: 1016.3 },
  qualityControlProcess: 'V020',
  seaSurfaceTemperature: { quality: '9', value: 27.6 },
  sections: [ 'AA2', 'SA1', 'MW1', 'AG1', 'GF1' ],
  skyCondition: {
    cavok: 'N',
    ceilingHeight: { determination: 'C', quality: '1', value: 6900 }
  },
  skyConditionObservation: {
    highCloudGenus: { quality: '1', value: '05' },
    lowCloudGenus: { quality: '9', value: '03' },
    lowestCloudBaseHeight: { quality: '9', value: 150 },
    lowestCloudCoverage: { quality: '1', value: '05' },
    midCloudGenus: { quality: '9', value: '08' },
    totalCoverage: { opaque: '99', quality: '1', value: '06' }
  },
  skyCoverLayer: {
    baseHeight: { quality: '9', value: 99999 },
    cloudType: { quality: '9', value: '05' },
    coverage: { quality: '1', value: '04' }
  },
  st: 'x+35700-027900',
  type: 'SAO',
  visibility: {
    distance: { quality: '1', value: 4000 },
    variability: { quality: '1', value: 'N' }
  },
  waveMeasurement: {
    method: 'I',
    seaState: { code: '99', quality: '9' },
    waves: { height: 99.9, period: 14, quality: '9' }
  },
  wind: {
    direction: { angle: 280, quality: '9' },
    speed: { quality: '1', rate: 30.3 },
    type: '9'
  }
}

Nota

Lo anterior es un ejemplo representativo. Los datos en transmisión no son estáticos, y cada usuario ve documentos distintos.

Volver

$validate

$lookup