AWS 클라우드 클라우드로 데이터 스트림 내보내기(콘솔)

AWS IoT Greengrass Core SDK for Python v1.5.0 이상을 다운로드하십시오.

다운로드한 패키지의 압축을 풀어 SDK를 가져옵니다. SDK는 greengrasssdk 폴더입니다.

Lambda 함수 배포 패키지에 SDK와 함께 포함시킬 패키지 종속성을 설치합니다.

transfer_stream.py이라는 로컬 파일에 다음과 같은 Python 코드 함수를 저장합니다.

import asyncio
import logging
import random
import time

from greengrasssdk.stream_manager import (
    ExportDefinition,
    KinesisConfig,
    MessageStreamDefinition,
    ReadMessagesOptions,
    ResourceNotFoundException,
    StrategyOnFull,
    StreamManagerClient,
)


# This example creates a local stream named "SomeStream".
# It starts writing data into that stream and then stream manager automatically exports  
# the data to a customer-created Kinesis data stream named "MyKinesisStream". 
# This example runs forever until the program is stopped.

# The size of the local stream on disk will not exceed the default (which is 256 MB).
# Any data appended after the stream reaches the size limit continues to be appended, and
# stream manager deletes the oldest data until the total stream size is back under 256 MB.
# The Kinesis data stream in the cloud has no such bound, so all the data from this script is
# uploaded to Kinesis and you will be charged for that usage.


def main(logger):
    try:
        stream_name = "SomeStream"
        kinesis_stream_name = "MyKinesisStream"

        # Create a client for the StreamManager
        client = StreamManagerClient()

        # Try deleting the stream (if it exists) so that we have a fresh start
        try:
            client.delete_message_stream(stream_name=stream_name)
        except ResourceNotFoundException:
            pass

        exports = ExportDefinition(
            kinesis=[KinesisConfig(identifier="KinesisExport" + stream_name, kinesis_stream_name=kinesis_stream_name)]
        )
        client.create_message_stream(
            MessageStreamDefinition(
                name=stream_name, strategy_on_full=StrategyOnFull.OverwriteOldestData, export_definition=exports
            )
        )

        # Append two messages and print their sequence numbers
        logger.info(
            "Successfully appended message to stream with sequence number %d",
            client.append_message(stream_name, "ABCDEFGHIJKLMNO".encode("utf-8")),
        )
        logger.info(
            "Successfully appended message to stream with sequence number %d",
            client.append_message(stream_name, "PQRSTUVWXYZ".encode("utf-8")),
        )

        # Try reading the two messages we just appended and print them out
        logger.info(
            "Successfully read 2 messages: %s",
            client.read_messages(stream_name, ReadMessagesOptions(min_message_count=2, read_timeout_millis=1000)),
        )

        logger.info("Now going to start writing random integers between 0 and 1000 to the stream")
        # Now start putting in random data between 0 and 1000 to emulate device sensor input
        while True:
            logger.debug("Appending new random integer to stream")
            client.append_message(stream_name, random.randint(0, 1000).to_bytes(length=4, signed=True, byteorder="big"))
            time.sleep(1)

    except asyncio.TimeoutError:
        logger.exception("Timed out while executing")
    except Exception:
        logger.exception("Exception while running")


def function_handler(event, context):
    return


logging.basicConfig(level=logging.INFO)
# Start up this sample code
main(logger=logging.getLogger())

다음 항목을 transfer_stream_python.zip라는 파일로 압축합니다. 이것이 Lambda 함수 배포 패키지입니다.

zip 파일을 생성할 때 이러한 항목만 포함시키고 포함 폴더는 포함시키지 않습니다.

먼저, Lambda 함수를 생성합니다.

1. AWS Management Console에서 [Services]를 선택한 다음 AWS Lambda 콘솔을 엽니다.

2. 함수 생성을 선택한 다음 새로 작성을 선택합니다.

3. 기본 정보 섹션에서 다음 값을 사용합니다.

   • [함수 이름]에 TransferStream을 입력합니다.

   • 실행 시간에서 Python 3.7을 선택합니다.

   • 권한의 경우, 기본 설정을 유지합니다. 이를 통해 기본 Lambda 권한을 부여하는 실행 역할을 생성합니다. 이 역할은 AWS IoT Greengrass에서 사용되지 않습니다.

4. 페이지 하단에서 함수 생성을 선택합니다.

이제 핸들러를 등록하고 Lambda 함수 배포 패키지를 업로드합니다.

1. 코드 탭의 코드 소스에서 다음에서 업로드를 선택합니다. 드롭다운에서 .zip 파일을 선택합니다.

   

2. 업로드를 선택한 다음 transfer_stream_python.zip 배포 패키지를 선택합니다. 그런 다음 저장을 선택합니다.

3. 함수의 코드 탭에 있는 런타임 설정에서 편집을 선택하고 다음 값을 입력합니다.

   • 실행 시간에서 Python 3.7을 선택합니다.

   • 핸들러에 transfer_stream.function_handler를 입력합니다.

4. Save를 선택합니다.

   참고

   AWS Lambda 콘솔의 테스트 버튼은 이 함수와 함께 작동하지 않습니다. AWS IoT Greengrass 코어 SDK에는 AWS Lambda 콘솔에서 Greengrass Lambda 함수를 독립적으로 실행하는 데 필요한 모듈이 포함되어 있지 않습니다. 이러한 모듈(예: greengrass_common)은 Greengrass 코어에 배포된 후 함수에 제공됩니다.

이제 Lambda 함수의 첫 번째 버전을 게시하고 버전의 별칭을 생성합니다.

1. [Actions] 메뉴에서 [Publish new revision]을 선택합니다.

2. 버전 설명에 First version을 입력한 후 게시를 선택합니다.

3. TransferStream: 1 구성 페이지의 작업 메뉴에서 별칭 생성을 선택합니다.

4. [Create a new alias] 페이지에서 다음 값을 사용합니다.

   • 이름에 GG_TransferStream을 입력합니다.

   • 버전에서 1을 선택합니다.

   참고

   AWS IoT Greengrass은(는) $LATEST 버전에서 Lambda 별칭을 지원하지 않습니다.

5. 생성을 선택합니다.

AWS IoT 콘솔 탐색 창의 관리에서 Greengrass 디바이스를 확장한 다음 그룹(V1)을 선택합니다.

대상 그룹을 선택합니다.

그룹 구성 페이지에서 Lambda 함수 탭을 선택합니다.

내 Lambda 함수에서 추가를 선택합니다.

Lambda 함수 추가 페이지에서 Lambda 함수에 대한 Lambda 함수를 선택합니다.

Lambda 버전의 경우 Alias:GG_TransferStream을 선택합니다.

이제 Greengrass 그룹에서 Lambda 함수의 동작을 결정하는 속성을 구성합니다.

Lambda 함수 구성 섹션에서 다음과 같이 변경합니다.

Lambda 함수 추가를 선택합니다.

그룹 구성 페이지에서 Lambda 함수 탭을 선택합니다.

시스템 Lambda 함수에서 스트림 관리자를 선택하고 상태를 확인합니다. 비활성화된 경우 편집을 선택합니다. 그런 다음 활성화 및 저장을 선택합니다. 이 튜토리얼의 기본 파라미터 설정을 사용할 수 있습니다. 자세한 내용은 AWS IoT Greengrass 스트림 관리자 구성 섹션을 참조하세요.

로컬 로그 구성에서 로컬 로깅이 구성되어 있는지 확인합니다.

Greengrass 시스템 구성 요소 또는 사용자 정의 Lambda 함수에 대해 로그가 구성되지 않은 경우에는 편집을 선택합니다.

사용자 Lambda 함수 로그 수준 및 Greengrass 시스템 로그 수준을 선택합니다.

로깅 수준 및 디스크 공간 제한을 기본값으로 유지한 후 저장을 선택합니다.

AWS IoT Greengrass 코어가 실행 중인지 확인합니다. 필요한 경우 Raspberry Pi 터미널에서 다음 명령을 실행합니다.

1. 대몬(daemon)이 실행 중인지 확인하려면:

   ps aux | grep -E 'greengrass.*daemon'

   출력에 root에 대한 /greengrass/ggc/packages/ggc-version/bin/daemon 입력이 포함되어 있는 경우에는 대몬(daemon)이 실행 중인 것입니다.

   참고

   경로의 버전은 코어 디바이스에 설치된 AWS IoT Greengrass 코어 소프트웨어 버전에 따라 다릅니다.

2. 대몬(daemon)을 시작하려면:

   cd /greengrass/ggc/core/
   sudo ./greengrassd start

그룹 구성 페이지에서 배포를 선택합니다.

1. Lambda 함수 탭의 시스템 Lambda 함수 섹션에서 IP 감지기를 선택하고 편집을 선택합니다.

2. IP 감지기 설정 편집 대화 상자에서 MQTT 브로커 엔드포인트 자동 탐지 및 재정의를 선택합니다.

3. Save를 선택합니다.

   이렇게 하면 디바이스가 IP 주소, DNS, 포트 번호 등 코어의 연결 정보를 자동으로 획득할 수 있습니다. 자동 탐지가 바람직하지만 AWS IoT Greengrass은(는) 수동으로 지정된 엔드포인트도 지원합니다. 그룹이 처음 배포될 때만 검색 방법 메시지가 표시됩니다.

   참고

   메시지가 표시되면 Greengrass 서비스 역할을 생성할 권한을 부여하고 이 권한을 현재 AWS 리전의 AWS 계정과 연결합니다. 이 역할은 AWS IoT Greengrass가 AWS 서비스의 리소스에 액세스할 수 있습니다.

   배포 페이지에 배포 타임스탬프, 버전 ID, 상태가 표시됩니다. 완료되면 배포에 대해 성공적으로 완료했습니다 상태가 표시되어야 합니다.

   문제 해결에 대한 도움말은 AWS IoT Greengrass 문제 해결 섹션을 참조하십시오.

Amazon Kinesis 콘솔의 Kinesis Data Streams에서 MyKinesisStream을 선택합니다.

MyKinesisStream 페이지에서 모니터링을 선택합니다. 테스트가 성공적이면 Put Records(레코드 넣기) 차트에 데이터가 표시되어야 합니다. 연결에 따라 데이터가 표시되기까지 1분 정도 걸릴 수 있습니다.

cd /greengrass/ggc/core/
sudo ./greengrassd stop

코어에서 TransferStream Lambda 함수를 제거합니다.

1. AWS IoT 콘솔 탐색 창의 관리에서 Greengrass 디바이스를 확장한 다음 그룹(V1)을 선택합니다.

2. Greengrass 그룹에서 사용자 그룹을 선택합니다.

3. Lambdas 페이지에서 TransferStream 함수에 줄임표(...)를 선택한 다음 함수 제거를 선택합니다.

4. 작업에서 배포를 선택합니다.