Amazon Kinesis - Amazon Timestream
Amazon Managed Service for Apache Flink 사용하기EventBridge 파이프 사용

Amazon Timestream for LiveAnalytics와 유사한 기능을 원하는 경우 Amazon Timestream for InfluxDB를 고려해 보세요. 간소화된 데이터 수집과 실시간 분석을 위한 10밀리초 미만의 쿼리 응답 시간을 제공합니다. 여기에서 자세히 알아보세요.

Amazon Kinesis

Managed Service for Apache Flink용 샘플 Timestream 데이터 커넥터를 사용하여 Kinesis Data Streams에서 Timestream for LiveAnalytics로 데이터를 전송할 수 있습니다. 자세한 내용은 Amazon Managed Service for Apache Flink for Apache Flink를 참조하세요.

EventBridge 파이프를 사용하여 Timestream으로 Kinesis 데이터 전송

EventBridge 파이프를 사용하여 Kinesis 스트림에서 Amazon Timestream for LiveAnalytics 테이블로 데이터를 전송할 수 있습니다.

파이프는 지원되는 소스와 대상 간의 지점 간 통합을 위한 것으로 고급 변환 및 강화를 지원합니다. 파이프는 이벤트 중심 아키텍처를 개발할 때 전문 지식 및 통합 코드의 필요성을 줄여줍니다. 파이프를 설정하려면 소스를 선택하고, 선택적 필터링을 추가하고, 선택적 보강을 정의하고, 이벤트 데이터의 대상을 선택합니다.

소스는 이벤트를 EventBridge 파이프로 전송하고, 파이프는 일치하는 이벤트를 필터링하여 대상으로 라우팅합니다.

이 통합을 통해 Timestream의 시계열 데이터 분석 기능의 성능을 활용하는 동시에 데이터 수집 파이프라인을 간소화할 수 있습니다.

Timestream과 함께 EventBridge 파이프를 사용하면 다음과 같은 이점이 있습니다.

  • 실시간 데이터 수집: Kinesis에서 Timestream for LiveAnalytics로 직접 데이터를 스트리밍하여 실시간 분석 및 모니터링을 활성화합니다.

  • 원활한 통합: EventBridge 파이프를 활용하여 복잡한 사용자 지정 통합 없이 데이터 흐름을 관리합니다.

  • 향상된 필터링 및 변환: 특정 데이터 처리 요구 사항을 충족하기 위해 Timestream에 저장되기 전에 Kinesis 레코드를 필터링하거나 변환합니다.

  • 확장성: 처리량이 많은 데이터 스트림을 처리하고 내장된 병렬 처리 및 배치 처리 기능으로 효율적인 데이터 처리를 보장합니다.

구성

Kinesis에서 Timestream로 데이터를 스트리밍하도록 EventBridge 파이프를 설정하려면 다음 단계를 따르세요.

  1. Kinesis 스트림 생성

    데이터를 수집할 활성 Kinesis 데이터 스트림이 있는지 확인합니다.

  2. Timestream 데이터베이스 및 테이블 생성

    데이터가 저장될 Timestream 데이터베이스와 테이블을 설정합니다.

  3. EventBridge 파이프 구성:

    • 소스: Kinesis 스트림을 소스로 선택합니다.

    • 대상: 대상으로 Timestream을 선택합니다.

    • 배치 설정: 배치 기간과 배치 크기를 정의하여 데이터 처리를 최적화하고 지연 시간을 줄입니다.

중요

파이프를 설정할 때는 몇 개의 레코드를 수집하여 모든 구성의 정확성을 테스트하는 것이 좋습니다. 파이프를 성공적으로 생성하더라도 파이프라인이 올바르고 데이터가 오류 없이 흐른다는 보장은 없습니다. 매핑 적용 후 잘못된 테이블, 잘못된 동적 경로 파라미터 또는 잘못된 Timestream 레코드와 같은 런타임 오류가 있을 수 있으며, 이는 실제 데이터가 파이프를 통해 흐를 때 발견됩니다.

다음 구성은 데이터가 수집되는 속도를 결정합니다.

  • BatchSize: Timestream for LiveAnalytics로 전송할 배치의 최대 크기입니다. 범위: 0~100. 최대 처리량을 얻으려면 이 값을 100으로 유지하는 것이 좋습니다.

  • MaximumBatchingWindowInSeconds: 배치 크기가 채워질 때까지 기다리는 최대 시간으로, 이 시간이 지나면 배치가 Timestream for LiveAnalytics 대상으로 전송됩니다. 수신 이벤트 속도에 따라 이 구성은 수집 지연을 결정하므로 데이터를 거의 실시간으로 Timestream에 계속 전송하려면 이 값을 < 10초로 유지하는 것이 좋습니다.

  • ParallelizationFactor: 각 샤드에서 동시에 처리할 배치 수입니다. 최대 처리량과 거의 실시간 수집을 얻으려면 최대 값 10을 사용하는 것이 좋습니다.

    스트림을 여러 대상이 읽는 경우 향상된 팬아웃을 사용하여 파이프에 전용 소비자를 제공하여 높은 처리량을 달성합니다. 자세한 내용은 Kinesis Data Streams 사용 설명서의 Kinesis Data Streams API를 사용하여 향상된 팬아웃 소비자 개발을 참조하세요.

참고

달성할 수 있는 최대 처리량은 계정당 동시 파이프 실행 횟수로 제한됩니다.

다음 구성은 데이터 손실을 방지합니다.

  • DeadLetterConfig: 사용자 오류로 인해 이벤트를 Timestream for LiveAnalytics에 수집할 수 없는 경우 데이터 손실을 방지하도록 DeadLetterConfig를 항상 구성하는 것이 좋습니다.

다음 구성 설정을 사용하여 파이프의 성능을 최적화하면 레코드가 느려지거나 차단되는 것을 방지할 수 있습니다.

  • MaximumRecordAgeInSeconds: 이보다 오래된 레코드는 처리되지 않으며 DLQ로 직접 이동합니다. 이 값을 대상 Timestream 테이블의 구성된 메모리 스토어 보존 기간보다 높지 않도록 설정하는 것이 좋습니다.

  • MaximumRetryAttempts: 레코드가 DeadLetterQueue로 전송되기 전에 레코드에 대한 재시도 횟수입니다. 이를 10으로 구성하는 것이 좋습니다. 이렇게 하면 일시적인 문제를 해결하는 데 도움이 될 수 있으며, 지속적인 문제의 경우 레코드가 DeadLetterQueue로 이동하고 나머지 스트림의 차단을 해제합니다.

  • OnPartialBatchItemFailure: 부분 배치 처리를 지원하는 소스의 경우 이를 활성화하고 실패한 레코드를 삭제/DLQ로 전송하기 전에 추가로 다시 시도하도록 AUTOMATIC_BISECT로 구성하는 것이 좋습니다.

구성 예

다음은 Kinesis 스트림에서 Timestream 테이블로 데이터를 스트리밍하도록 EventBridge 파이프를 구성하는 방법의 예입니다.

예 Timestream에 대한 IAM 정책 업데이트
JSON
{
    "Version":"2012-10-17",		 	 	 
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "timestream:WriteRecords"
            ],
            "Resource": [
                "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "timestream:DescribeEndpoints"
            ],
            "Resource": "*"
        }
    ]
}
예 Kinesis 스트림 구성
{
  "Source": "arn:aws:kinesis:us-east-1:123456789012:stream/my-kinesis-stream",
  "SourceParameters": {
    "KinesisStreamParameters": {
        "BatchSize": 100,
        "DeadLetterConfig": {
            "Arn": "arn:aws:sqs:us-east-1:123456789012:my-sqs-queue"
        },
       "MaximumBatchingWindowInSeconds": 5,
        "MaximumRecordAgeInSeconds": 1800,
        "MaximumRetryAttempts": 10,
        "StartingPosition": "LATEST",
       "OnPartialBatchItemFailure": "AUTOMATIC_BISECT"
    }
  }
}
예 Timestream 대상 구성
{
    "Target": "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table",
    "TargetParameters": {
        "TimestreamParameters": {
            "DimensionMappings": [
                {
                    "DimensionName": "sensor_id",
                    "DimensionValue": "$.data.device_id",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_type",
                    "DimensionValue": "$.data.sensor_type",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_location",
                    "DimensionValue": "$.data.sensor_loc",
                    "DimensionValueType": "VARCHAR"
                }
            ],
            "MultiMeasureMappings": [
                {
                    "MultiMeasureName": "readings",
                    "MultiMeasureAttributeMappings": [
                        {
                            "MultiMeasureAttributeName": "temperature",
                            "MeasureValue": "$.data.temperature",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "humidity",
                            "MeasureValue": "$.data.humidity",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "pressure",
                            "MeasureValue": "$.data.pressure",
                            "MeasureValueType": "DOUBLE"
                        }
                    ]
                }
            ],
            "SingleMeasureMappings": [],
            "TimeFieldType": "TIMESTAMP_FORMAT",
            "TimestampFormat": "yyyy-MM-dd HH:mm:ss.SSS",
            "TimeValue": "$.data.time",
            "VersionValue": "$.approximateArrivalTimestamp"
        }
    }
}

이벤트 변환

EventBridge 파이프를 사용하면 Timestream에 도달하기 전에 데이터를 변환할 수 있습니다. 변환 규칙을 정의하여 필드 이름 변경과 같은 수신 Kinesis 레코드를 수정할 수 있습니다.

Kinesis 스트림에 온도 및 습도 데이터가 포함되어 있다고 가정합니다. EventBridge 변환을 사용하여에 삽입하기 전에 이러한 필드의 이름을 바꿀 수 있습니다Timestream.

모범 사례

배치 처리 및 버퍼링

  • 쓰기 지연 시간과 처리 효율성 간의 균형을 맞추도록 배치 처리 기간과 크기를 구성합니다.

  • 배치 처리 기간을 사용하여 처리 전에 충분한 데이터를 누적하여 자주 발생하는 작은 배치의 오버헤드를 줄입니다.

병렬 처리

ParallelizationFactor 설정을 활용하여 특히 처리량이 많은 스트림의 경우 동시성을 높입니다. 이렇게 하면 각 샤드의 여러 배치를 동시에 처리할 수 있습니다.

데이터 변환

에 저장하기 전에 EventBridge 파이프의 변환 기능을 활용하여 레코드를 필터링하고 개선합니다Timestream. 이렇게 하면 데이터를 분석 요구 사항에 맞추는 데 도움이 될 수 있습니다.

보안.

  • EventBridge 파이프에 사용되는 IAM 역할에 Kinesis에서 읽고 Timestream에 쓰는 데 필요한 권한이 있는지 확인합니다.

  • 암호화 및 액세스 제어 조치를 사용하여 전송 중 및 저장 데이터를 보호합니다.

실패 디버깅

  • 파이프 자동 비활성화

    대상이 없거나 권한 문제가 있는 경우 약 2시간 내에 파이프가 자동으로 비활성화됩니다.

  • 제한

    파이프에는 스로틀이 줄어들 때까지 자동으로 백오프하고 재시도하는 기능이 있습니다.

  • 로그 활성화

    오류 수준에서 로그를 활성화하고 실행 데이터를 포함하여 실패에 대한 더 많은 인사이트를 얻는 것이 좋습니다. 실패 시 이러한 로그에는 Timestream에서 전송/수신된 요청/응답이 포함됩니다. 이렇게 하면 관련 오류를 이해하고 필요한 경우 레코드를 수정한 후 다시 처리할 수 있습니다.

모니터링

데이터 흐름 문제를 탐지하려면 다음에 대한 경보를 설정하는 것이 좋습니다.

  • 소스에서 레코드의 최대 기간

    • GetRecords.IteratorAgeMilliseconds

  • 파이프의 실패 지표

    • ExecutionFailed

    • TargetStageFailed

  • Timestream API 오류 쓰기

    • UserErrors

추가 모니터링 지표는 EventBridge 사용 설명서의 EventBridge 모니터링을 참조하세요.