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

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

Amazon Kinesis

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

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

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

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

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

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

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

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

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

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

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

구성

Kinesis에서 로 데이터를 스트리밍하도록 EventBridge 파이프를 설정하려면 다음 단계를 Timestream수행합니다.

  1. Kinesis 스트림 생성

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

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

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

  3. EventBridge 파이프 구성:

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

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

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

중요

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

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

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

  • MaximumBatchingWindowInSeconds: 배치가 LiveAnalytics 대상의 Timestream으로 전송되기 전에 batchSize를 채우기 위해 대기하는 최대 시간입니다. 수신되는 이벤트의 속도에 따라이 구성은 수집 지연을 결정하므로 데이터를 거의 실시간으로에 계속 전송하려면이 값을 < 10초 Timestream 로 유지하는 것이 좋습니다.

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

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

참고

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

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

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

다음 구성 설정을 사용하여 파이프의 성능을 최적화하면 레코드가 속도 저하 또는 차단을 일으키지 않도록 방지할 수 있습니다.

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

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

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

구성 예

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

예 IAM 에 대한 정책 업데이트 Timestream
{
    "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 Pipes의 변환 기능을 활용하여 레코드를 저장하기 전에 필터링하고 개선합니다 Timestream. 이를 통해 데이터를 분석 요구 사항에 맞게 조정할 수 있습니다.

보안

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

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

디버깅 실패

  • 파이프 자동 비활성화

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

  • 제한

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

  • 로그 활성화

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

모니터링

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

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

    • GetRecords.IteratorAgeMilliseconds

  • 파이프의 실패 지표

    • ExecutionFailed

    • TargetStageFailed

  • Timestream API 오류 쓰기

    • UserErrors

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