.sp. 프로세스() (mongosh 메서드)

정의

sp.process()

Creates an ephemeral Stream Processor on the current Stream Processing Workspace.

호환성

이 메서드는 Atlas Stream Processing Workspaces에서 지원됩니다.

구문

The sp.process() method has the following syntax:

sp.process(
  [
    <pipeline>
  ],
  {
    <options>
  }
)

명령 필드

sp.process() 는 다음 필드를 사용합니다.

필드	유형	필요성	설명
`pipeline`	배열	필수 사항	Stream aggregation pipeline the stream processor applies to your streaming data.
`options`	객체	옵션	스트림 프로세서에 대한 선택 사항을 정의하는 객체입니다.
`options.limit`	integer	옵션	터미널로 반환할 문서의 최대 개수입니다. `sp.process()`은 이 개수의 문서를 반환한 후 세션을 종료합니다.
`options.tier`	문자열	옵션	Stream Processing에 할당할 계층입니다. 계층 이름은 대소문자를 구분하지 않습니다. 이 필드를 생략하면 Stream Processing Workspace의 기본 계층을 사용합니다. 다음 값 중 하나여야 합니다. `SP2` `SP5` `SP10` `SP30` `SP50` To learn more, see Tiers.

행동

sp.process() 현재 Stream Processing 작업 공간에 임시적인 이름 없는 스트림 프로세서를 생성하고 즉시 초기화합니다. 이 Stream Processor는 실행 중에만 유지됩니다. 임시 Stream Processor를 종료하는 경우 사용하려면 다시 생성해야 합니다.

일시적 스트림 프로세서는 다음 중 하나라도 발생하면 중지됩니다.

프로세서는 options.limit에 설정된 문서 수를 반환합니다.
10 minutes elapse.
프로세서를 중지합니다.

액세스 제어

를 실행 sp.process() 사용자는 atlasAdmin 역할 있어야 합니다.

예시

다음 예시 연결에서 데이터를 수집하는 임시 스트림 프로세서를 sample_stream_solar 만듭니다. 프로세서는 필드 값이 인 모든 문서를 제외하고 device_id device_8 나머지는 10지속 시간이 초인 텀블링 창 으로 전달합니다. 각 창 수신한 문서를 그룹화한 다음 각 그룹 의 다양한 통계를 반환합니다. 그런 다음 스트림 프로세서는 solar_db.solar_coll mongodb1 연결을 통해 이러한 레코드를 에 병합합니다.

sp.process(
  [
    {
      $source: {
        connectionName: 'sample_stream_solar',
        timeField: {
          $dateFromString: {
            dateString: '$timestamp'
          }
        }
      }
    },
    {
      $match: {
        $expr: {
          $ne: [
            "$device_id",
            "device_8"
          ]
        }
      }
    },
    {
      $tumblingWindow: {
        interval: {
          size: Int32(10),
          unit: "second"
        },
        "pipeline": [
          {
            $group: {
              "_id": { "device_id": "$device_id" },
              "max_temp": { $max: "$obs.temp" },
              "max_watts": { $max: "$obs.watts" },
              "min_watts": { $min: "$obs.watts" },
              "avg_watts": { $avg: "$obs.watts" },
              "median_watts": {
                $median: {
                  input: "$obs.watts",
                  method: "approximate"
                }
              }
            }
          }
        ]
      }
    },
    {
      $merge: {
        into: {
          connectionName: "mongodb1",
          db: "solar_db",
          coll: "solar_coll"
        },
        on: ["_id"]
      }
    }
  ]
)

자세히 알아보기

돌아가기

sp.listWorkspaceDefaults

sp.processor.drop