/ /

`$https` Estágio de agregação (processamento de fluxo)

Definição

$https

O estágio $https especifica uma conexão no Registro de Conexão para enviar solicitações HTTPS. Cada vez que um estágio anterior passa um documento para $https, o estágio envia uma nova solicitação.

Colocação

$https deve vir após o estágio $source, pode vir antes do estágio $emit ou $merge, ou pode servir como o estágio final de um pipeline. Você pode usar $https nos pipelines internos do Windows do processador de fluxo.

Sintaxe

Para enviar uma solicitação de HTTPS para uma conexão específica:

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

O estágio $https recebe um documento com os seguintes campos:

Campo	Tipo	necessidade	Descrição
`connectionName`	string	Obrigatório	Rótulo que identifica a conexão no Registro de Conexão para enviar uma solicitação `HTTPS`.
`path`	corda \| expressão	Opcional	Caminho a ser anexado ao URL ao qual seu `connectionName` se resolve. Por exemplo, se você especificar um `connectionName` que se resolve em `https://sample.com`, você pode especificar um `path` de `"endpoint"` para que seu processador de fluxo envie solicitações `HTTPS` para `https://sample.com/endpoint`. Se você definir `path` como uma expressão, essa expressão deve ser avaliada como uma string. O ponto de extremidade da API que você chama deve ser idempotente.
`parameters`	documento	Opcional	Documento contendo pares de valores-chave para passar como parâmetros para sua chamada de ponto de conexão da API. Cada chave deve ser uma string e cada valor deve avaliar para um valor numérico, de string ou booleano. Este campo suporta expressões como valores.
`method`	string	Opcional	Método de solicitação HTTPS para sua conexão. Deve ser um dos seguintes valores: `"GET"` `"POST"` `"PUT"` `"PATCH"` `"DELETE"` Padrão é `"GET"`.
`headers`	documento	Opcional	Documento contendo pares de valores-chave para passar como cabeçalhos para o ponto de conexão da API. Cada chave deve ser uma string e cada valor deve avaliar para uma string. Este campo suporta expressões como valores. Se o endpoint da API exigir autenticação, como uma chave de API ou autenticação por Token de Acesso Bearer, você deve adicionar os detalhes de autenticação como cabeçalhos ao definir a conexão para evitar fornecê-los como texto simples como parte deste operador. Nomes e valores de cabeçalhos HTTP inválidos não são enviados ao ponto de extremidade da API. Em vez disso, estes são ignorados. Para saber mais sobre cabeçalhos HTTP inválidos, consulte RFC 9110. Se uma expressão em um valor falhar ou for avaliada como um tipo diferente de string, a mensagem será enviada para a fila de mensagens não entregues (DLQ) e o operador não enviará essa solicitação para o ponto de extremidade da API.
`as`	string	Condicional	Nome do campo para a resposta da REST API. O parâmetro `as` é necessário quando usado como um estágio não coletor (intermediário) e opcional quando usado como o último estágio ou estágio coletor. Se o ponto de extremidade retornar 0 bytes, o operador não configurará o campo `as`. O operador suporta respostas com um `Content-Type` de `application/json` ou `text/plain`. Se o ponto de extremidade da API retornar uma resposta com um `Content-Type` diferente, o operador processará o documento com base no comportamento `onError` que você definir. Se o ponto de extremidade da API retornar uma resposta sem um `Content-Type` definido, o operador assume que a resposta é `application/json`.
`onError`	string	Opcional	Comportamento quando o operador encontra uma falha relacionada a `HTTPS`. Deve ser um dos seguintes valores: `"dlq"` : Passe o documento afetado para a fila de mensagens não entregues (DLQ). `"ignore"` : ignore o erro e passe o documento afetado para a próxima etapa do pipeline. `"fail"` : Finalizar o processador de fluxo em caso de erro. O operador considera todos os códigos de status HTTP `2XX` como bem-sucedidos. Se o operador receber algum dos seguintes códigos de status HTTP em resposta, ele seguirá o comportamento com base no valor que você fornecer neste campo: `400` `404` `410` `413` `414` `431` O operador considera qualquer outro código de status HTTP como um erro `"fail"`. Por exemplo, se o ponto de extremidade da API retornar um código de status HTTP `500`, o processador entra em um estado de falha e para. `onError` não é acionado por erros decorrentes da configuração incorreta do operador `$https` em si, como expressões inválidas. Padrão é `"dlq"`.
`payload`	array	Opcional	Pipeline interno personalizado que permite personalizar o corpo da solicitação enviado ao ponto de extremidade da API. `payload` suporta as seguintes expressões: `$project` `$addFields` `$replaceRoot` `$set` Por padrão, a mensagem inteira é enviada para o endpoint da API. O Atlas Stream Processing envia uma carga útil JSON de modo relaxado para o endpoint da API. Corpos de solicitação HTTP inválidos não são enviados para o endpoint da API. Em vez disso, eles são enviados para a fila de mensagens não entregues (DLQ). Para saber mais sobre corpos de solicitação HTTP inválidos, consulte RFC 9110.
`config`	documento	Opcional	documento que contém campo que substituem vários valores padrão.
`config.connectionTimeoutSec`	inteiro	Opcional	O tempo, em segundos, após o qual uma tentativa de conexão `HTTPS` atinge o tempo limite se não for possível estabelecer uma conexão. Padrão é `30`.
`config.parseJsonStrings`	booleano	Opcional	Configuração que determina se o Atlas itera recursivamente pela resposta do servidor e serializa valores de string que contêm JSON válido (usando aspas escapadas) em bson válido para que você possa trabalhar com os resultados nos estágios do pipeline downstream. Padrão é `false`.
`config.requestTimeoutSec`	inteiro	Opcional	O tempo, em segundos, após o qual uma solicitação `HTTPS` atinge o tempo limite se nenhuma resposta for recebida. Padrão é `60`.

Exemplos

Uma fonte de dados de streaming gera relatórios meteorológicos detalhados a partir de vários locais, em conformidade com o esquema do Conjunto de dados meteorológicos de amostra. A seguinte agregação tem três estágios:

O estágio $source estabelece uma conexão com o intermediário do Apache Kafka coletando esses relatórios em um tópico chamado my_weatherdata, expondo cada registro à medida que ele é ingerido para os estágios de agregação subsequentes. Esse estágio também substitui o nome do campo de carimbo de data/hora que ele projeta, definindo-o como ingestionTime.
Para cada registro do intermediário Apache Kafka, o estágio $https envia uma solicitação para uma fonte meteorológica HTTPS definida na conexão https_weather. A solicitação usa o position.coordinates do registro na solicitação HTTPS para reunir uma previsão de alta temperatura de sete dias em graus Celsius para esse local, que é adicionado ao documento de pipeline em um campoairTemperatureForecast.
O estágio $merge escreve o resultado em uma collection do Atlas chamada stream no banco de dadossample_weatherstream. Se não existir tal banco de dados ou coleção, o Atlas os criará.

{
  '$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 visualizar os documentos na coleção sample_weatherstream.stream resultante, conecte-se ao cluster Atlas e execute o seguinte 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'
  }
}

Observação

O exemplo anterior é representativo. Os dados de streaming não são estáticos e cada usuário vê documentos distintos.

Voltar

$validar

$lookup

$https Estágio de agregação (processamento de fluxo)

Definição

Colocação

Sintaxe

Exemplos

`$https` Estágio de agregação (processamento de fluxo)