Atlas Stream Processing 시작하기

Atlas UI 에서 스트림 프로세서를 만들려면 Atlas 프로젝트 의 Stream Processing 페이지로 고 (Go) Stream Processing 인스턴스 창에서 Configure 를 클릭합니다. 그런 다음 시각적 빌더 또는 JSON 편집기 중 하나를 사용하여 solarDemo라는 스트림 프로세서를 구성합니다.

mongosh 에서 다음 명령을 실행하여 solarDemo라는 영구 스트림 프로세서를 생성합니다.

1. Atlas Stream Processing 인스턴스에 연결합니다.

   Atlas Stream Processing 인스턴스와 연결된 연결 string 을 사용하여 mongosh을(를) 사용하여 연결합니다.

   1. Atlas Stream Processing 인스턴스의 창에서 Connect 을(를) 클릭합니다.

   2. Connect to your instance 대화 상자에서 Shell 탭 선택합니다.

   3. 대화 상자에 표시된 연결 문자열 복사합니다. 형식은 다음과 같으며, 여기서 <atlas-stream-processing-url> 은 스트림 처리 인스턴스 의 URL 이고 <username> 은 atlasAdmin 역할 가진 데이터베이스 사용자의 사용자 이름 입니다.

      mongosh "mongodb://<atlas-stream-processing-url>/"
      --tls --authenticationDatabase admin --username <username>
      --password <password>

   4. 연결 문자열 터미널에 붙여넣고 <password> 자리 표시자를 사용자의 자격 증명 으로 바꿉니다.

      Enter 키를 눌러 실행 하고 Stream Processing 인스턴스 에 연결합니다.

2. $source 단계를 구성합니다.

   sample_stream_solar 소스에서 데이터를 수집하는 $source 단계에 대한 변수를 정의합니다.

   let s = {
     $source: {
       connectionName: "sample_stream_solar"
     }
   }

3. $group 단계를 구성합니다.

   group_id에 따라 각 태양광 발전 장치의 최대 온도와 평균, 최대 및 최소 와트수를 파생하는 $group 단계에 대한 변수를 정의합니다.

   let g = {
     $group: {
       _id: "$group_id",
       max_temp: {
         $max: "$obs.temp"
       },
       avg_watts: {
          $avg: "$obs.watts"
       },
       max_watts: {
         $max: "$obs.watts"
       },
       min_watts: {
         $min: "$obs.watts"
       }
     }
   }

4. $tumblingWindow 단계를 구성합니다.

   스트리밍 데이터에 대해 $group 와 같은 축적을 수행하기 위해 Atlas Stream Processing은 Windows 를 사용하여 데이터 세트를 바인딩합니다. 스트림 연속적인 10초 간격으로 분리하는 $tumblingWindow 단계에 대한 변수를 정의합니다.

   예시 $group 단계에서 max_watts 값을 계산할 때 이전 10 초 동안 특정 group_id 가 수집된 모든 문서의 obs.watts 값에서 최대값을 추출합니다.

    let t = {
      $tumblingWindow: {
        interval: {
          size: NumberInt(10),
          unit: "second"
        },
        pipeline: [g]
     }
   }

5. $merge 단계를 구성합니다.

   연결된 Atlas cluster 의 solarDb 데이터베이스 에 있는 solarColl 컬렉션 에 처리된 스트리밍 데이터를 쓰는 $merge 단계에 대한 변수를 정의합니다.

   let m = {
     $merge: {
       into: {
         connectionName: "mongodb1",
         db: "solarDb",
         coll: "solarColl"
       }
     }
   }

6. 스트림 프로세서를 생성합니다.

   sp.createStreamProcessor() 메서드를 사용하여 새 스트림 프로세서에 이름을 할당하고 해당 집계 파이프라인 선언합니다. $group 단계는 $tumblingWindow의 중첩된 파이프라인 에 속하며, 프로세서 파이프라인 정의에 포함해서는 안 됩니다.

   sp.createStreamProcessor("solarDemo", [s, t, m])

   이는 solarDemo라는 스트림 프로세서를 생성하여 이전에 정의한 쿼리를 적용하고 처리된 데이터를 연결된 클러스터의 solarDb 데이터베이스의 solarColl 컬렉션에 기록합니다. 이 프로세서는 태양광 장치로부터의 관측 데이터를 10초 간격으로 나누어 도출한 다양한 측정값을 반환합니다.

   Atlas Stream Processing 미사용 데이터 데이터베이스에 쓰는 방법에 대해 자세히 학습하려면 $merge (스트림 처리)를 참조하세요.

스트림 처리 인스턴스 의 스트림 프로세서 목록에서 스트림 프로세서의 Start 아이콘을 클릭합니다.

sp.processor.start() 에서 메서드를 mongosh 사용합니다.

sp.solarDemo.start()

스트림 프로세서가 Atlas 클러스터에 데이터를 쓰고 있는지 확인하려면 다음을 수행하세요.

1. Atlas에서 프로젝트의 Clusters 페이지로 이동합니다.

   경고: 탐색 기능 개선 작업 진행 중

   현재 새롭고 향상된 탐색 환경을 출시하고 있습니다. 다음 단계가 Atlas UI 의 보기와 일치하지 않는 경우 미리 보기 설명서를 참조하세요.

   1. 아직 표시되지 않은 경우 탐색 표시줄의 Organizations 메뉴에서 원하는 프로젝트가 포함된 조직을 선택합니다.

   2. 아직 표시되지 않은 경우 탐색 표시줄의 Projects 메뉴에서 원하는 프로젝트를 선택합니다.

   3. 아직 표시되지 않은 경우 사이드바에서 Clusters를 클릭합니다.

      Clusters(클러스터) 페이지가 표시됩니다.

2. cluster의 Browse Collections 버튼을 클릭합니다.

   데이터 탐색기 가 표시됩니다.

3. MySolar 컬렉션을 확인하세요.

프로세서가 활성 상태인지 mongosh 확인하려면: sp.processor.stats()에서 메서드를 사용합니다.

sp.solarDemo.stats()

이 메서드는 solarDemo 스트림 프로세서의 작동 통계를 보고합니다.

의 메서드를 사용하여 터미널에서 처리된 sp.processor.sample() 문서의 샘플링을 반환할 수도 있습니다.mongosh

sp.solarDemo.sample()

{
  _id: 10,
  max_temp: 16,
  avg_watts: 232,
  max_watts: 414,
  min_watts: 73
}

스트림 처리 인스턴스의 스트림 프로세서 목록에서 스트림 프로세서의 Delete () 아이콘을 클릭합니다.

표시되는 확인 대화 상자에서 스트림 프로세서(solarDemo)의 이름을 입력하여 삭제 것인지 확인한 다음 Delete을(를) 클릭합니다.

에서 메서드를 sp.processor.drop() 사용하여 mongosh 을(를) 삭제합니다.solarDemo

sp.solarDemo.drop()

solarDemo 삭제를 확인하려면 sp.listStreamProcessors() 메서드를 사용하여 사용 가능한 모든 스트림 프로세서를 나열합니다.

sp.listStreamProcessors()