Atlas Stream Processing 시작하기

To create a stream processor in the Atlas UI, go to the Stream Processing page for your Atlas project and click Configure in the pane for your stream processing workspace. Then choose between using the visual builder or the JSON editor to configure a stream processor named solarDemo:

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

1. Connect to your stream processing workspace.

   Use the connection string associated with your stream processing workspace to connect using mongosh.

   1. In the pane for your stream processing workspace, click Connect.

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

   3. Copy the connection string displayed in the dialog. It has the following format, where <atlas-stream-processing-url> is the URL of your stream processing workspace and <username> is the username of a database user with the atlasAdmin role:

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

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

      Press Enter to run it and connect to your stream processing workspace.

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 (스트림 처리)를 참조하세요.

In the list of stream processors for your stream processing workspace, click the Start icon for your stream processor.

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

sp.solarDemo.start()

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

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

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

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

   3. 사이드바에서 Database 제목 아래의 Clusters를 클릭합니다.

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

2. Atlas에서 프로젝트의 Data Explorer 페이지로 이동합니다.

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

   2. 아직 표시되지 않은 경우 내비게이션 바의 Projects 메뉴에서 프로젝트를 선택합니다.

   3. 사이드바에서 Database 제목 아래의 Data Explorer를 클릭합니다.

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

   참고

   Clusters 페이지로 이동하여 Shortcuts 제목 아래의 Data Explorer 을 클릭할 수도 있습니다.

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
}

In the list of stream processors for your stream processing workspace, click the Delete () icon for your stream processor.

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

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

sp.solarDemo.drop()

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

sp.listStreamProcessors()