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 instance. Then choose between using the visual builder or the JSON editor to configure a stream processor named solarDemo:

Run the following commands in mongosh to create a persistent stream processor named solarDemo:

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

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

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

   2. In the Connect to your instance dialog, select the Shell tab.

   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 instance 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. Paste the connection string into your terminal and replace the <password> placeholder with the credentials for the user.

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

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

   Define a variable for a $source stage that ingests data from the sample_stream_solar source.

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

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

   Define a variable for a $group stage that derives the maximum temperature and the average, maximum, and minimum wattages of each solar device according to its group_id.

   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 단계를 구성합니다.

   In order to perform accumulations such as $group on streaming data, Atlas Stream Processing uses windows to bound the data set. Define a variable for a $tumblingWindow stage that separates the stream into consecutive 10-second intervals.

   This means, for example, that when the $group stage computes a value for max_watts, it extracts the maximum value from the obs.watts values for all documents with a given group_id ingested in the previous 10 seconds.

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

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

   Define a variable for a $merge stage that writes the processed streaming data to a collection named solarColl in the solarDb database of your connected Atlas cluster.

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

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

   Use the sp.createStreamProcessor() method to assign a name to your new stream processor and declare its aggregation pipeline. The $group stage belongs to the nested pipeline of the $tumblingWindow, and you must not include it in the processor pipeline definition.

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

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

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

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

Use the sp.processor.start() method in mongosh:

sp.solarDemo.start()

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

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

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

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

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

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

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

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

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

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

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

To verify that the processor is active, use the sp.processor.stats() method in mongosh:

sp.solarDemo.stats()

This method reports operational statistics of the solarDemo stream processor.

You can also use the sp.processor.sample() method in mongosh to return a sampling of processed documents in the terminal.

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 instance, click the Delete () icon for your stream processor.

In the confirmation dialog that appears, type the name of the stream processor (solarDemo) to confirm that you want to delete it, and then click Delete.

Use the sp.processor.drop() method in mongosh to drop solarDemo:

sp.solarDemo.drop()

To confirm that you have dropped solarDemo, use the sp.listStreamProcessors() method to list all your available stream processors:

sp.listStreamProcessors()