RDS source

출처: 문서

본문

RDS 소스

rds 소스는 Amazon Relational Database Service (Amazon RDS)와 Amazon Aurora 데이터베이스에서 변경 데이터 캡처(CDC)를 가능하게 해요. 데이터베이스 복제 로그를 사용해 INSERT, UPDATE, DELETE 같은 데이터베이스 이벤트를 수신할 수 있고, Amazon Simple Storage Service (Amazon S3)로의 RDS 내보내기를 활용한 초기 로드도 지원해요.

이 소스는 다음 데이터베이스 엔진을 지원해요:

• Aurora MySQL 및 Aurora PostgreSQL
• RDS MySQL 및 RDS PostgreSQL

이 소스는 Aurora/RDS에서 데이터를 수집하기 위한 두 가지 수집 옵션을 제공해요:

1. Export: Aurora/RDS에서 S3로 전체 초기 내보내기를 수행하면 Aurora/RDS 데이터베이스의 현재 상태가 초기 로드돼요.
2. Stream: 데이터베이스 복제 로그(MySQL binlog 또는 PostgreSQL WAL)에서 이벤트를 스트리밍해요.

사용 방법

다음 예제 파이프라인은 rds 소스를 지정해요. Aurora MySQL 클러스터에서 데이터를 수집해요:

version: "2"
rds-pipeline:
  source:
    rds:
      db_identifier: "my-rds-instance"
      engine: "aurora-mysql"
      database: "mydb"
      authentication:
        username: "myuser"
        password: "mypassword"
      s3_bucket: "my-export-bucket"
      s3_region: "us-west-2"
      s3_prefix: "rds-exports"
      export:
        kms_key_id: "arn:aws:kms:us-west-2:123456789012:key/12345678-1234-1234-1234-123456789012"
        export_role_arn: "arn:aws:iam::123456789012:role/rds-export-role"
      stream: true
      aws:
        region: "us-west-2"
        sts_role_arn: "arn:aws:iam::123456789012:role/my-pipeline-role"

구성 옵션

다음 표는 rds 소스의 구성 옵션을 설명해요.

aws

AWS 구성에서는 다음 옵션을 사용해요.

authentication

데이터베이스 인증에는 다음 옵션을 사용해요.

tables

데이터 캡처에 포함할 테이블을 지정하려면 다음 옵션을 사용해요.

export options

다음 옵션으로 RDS 내보내기 기능을 사용자 지정할 수 있어요.

tls options

다음 옵션으로 데이터베이스 연결의 TLS를 구성할 수 있어요.

노출되는 메타데이터 속성

rds 소스가 처리하는 각 이벤트에는 다음 메타데이터가 추가돼요. 이 메타데이터 속성에는 표현식 문법의 getMetadata 함수를 사용해 접근할 수 있어요.

• primary_key: 데이터베이스 레코드의 기본 키예요. 복합 기본 키가 있는 테이블에서는 값이 | 구분자로 연결돼요.
• event_timestamp: 데이터베이스 변경이 발생한 시각의 epoch 밀리초 값이에요. 내보내기 이벤트에서는 내보내기 시각을, 스트림 이벤트에서는 트랜잭션 커밋 시각을 나타내요.
• document_version: 문서 버전으로 사용하기 위해 이벤트 타임스탬프에서 생성된 long 정수 값이에요.
• opensearch_action: 이벤트를 OpenSearch로 보낼 때 사용할 벌크 작업이에요. 예를 들어 index 또는 delete가 있어요.
• change_event_type: 스트림 이벤트의 타입이에요. insert, update, delete 중 하나예요.
• table_name: 이벤트가 발생한 데이터베이스 테이블의 이름이에요.
• schema_name: 이벤트가 발생한 스키마의 이름이에요. MySQL의 경우 schema_name은 database_name과 같아요.
• database_name: 이벤트가 발생한 데이터베이스의 이름이에요.
• ingestion_type: 이벤트가 내보내기에서 발생했는지, 스트림에서 발생했는지를 나타내요. 유효한 값은 EXPORT와 STREAM이에요.
• s3_partition_key: 이벤트는 처리 전에 S3 스테이징 버킷에 저장돼요. 이 메타데이터는 처리 전에 이벤트가 저장되는 S3 버킷 내의 위치를 나타내요.

권한

RDS를 소스로 실행하는 데 필요한 권한은 다음과 같아요:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "allowReadingFromS3Buckets",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:DeleteObject",
        "s3:GetBucketLocation",
        "s3:ListBucket",
        "s3:PutObject"
      ],
      "Resource": [
        "arn:aws:s3:::s3_bucket",
        "arn:aws:s3:::s3_bucket/*"
      ]
    },
    {
      "Sid": "AllowDescribeInstances",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeDBInstances"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:db:*"
      ]
    },
    {
      "Sid": "AllowDescribeClusters",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeDBClusters"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:cluster:*"
      ]
    },
    {
      "Sid": "AllowSnapshots",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeDBClusterSnapshots",
        "rds:CreateDBClusterSnapshot",
        "rds:DescribeDBSnapshots",
        "rds:CreateDBSnapshot",
        "rds:AddTagsToResource"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:cluster:*",
        "arn:aws:rds:region:account-id:cluster-snapshot:*",
        "arn:aws:rds:region:account-id:db:*",
        "arn:aws:rds:region:account-id:snapshot:*"
      ]
    },
    {
      "Sid": "AllowExport",
      "Effect": "Allow",
      "Action": [
        "rds:StartExportTask"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:cluster:*",
        "arn:aws:rds:region:account-id:cluster-snapshot:*",
        "arn:aws:rds:region:account-id:snapshot:*"
      ]
    },
    {
      "Sid": "AllowDescribeExports",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeExportTasks"
      ],
      "Resource": "*"
    },
    {
      "Sid": "AllowAccessToKmsForExport",
      "Effect": "Allow",
      "Action": [
        "kms:Decrypt",
        "kms:Encrypt",
        "kms:DescribeKey",
        "kms:RetireGrant",
        "kms:CreateGrant",
        "kms:ReEncrypt*",
        "kms:GenerateDataKey*"
      ],
      "Resource": [
        "arn:aws:kms:region:account-id:key/export-key-id"
      ]
      },
    {
      "Sid": "AllowPassingExportRole",
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": [
        "arn:aws:iam::account-id:role/export-role"
      ]
    }
  ]
}

메트릭

rds 소스에는 다음 메트릭이 포함돼요:

• exportJobSuccess: 성공한 RDS 내보내기 작업 수예요.
• exportJobFailure: 실패한 RDS 내보내기 작업 수예요.
• exportS3ObjectsTotal: S3에서 발견된 내보내기 데이터 파일의 총 개수예요.
• exportS3ObjectsProcessed: S3에서 성공적으로 처리된 내보내기 데이터 파일의 총 개수예요.
• exportS3ObjectsErrors: S3에서 처리에 실패한 내보내기 데이터 파일의 총 개수예요.
• exportRecordsTotal: 내보내기에서 발견된 레코드의 총 개수예요.
• exportRecordsProcessed: 성공적으로 처리된 내보내기 레코드의 총 개수예요.
• exportRecordsProcessingErrors: 내보내기 레코드 처리 오류 수예요.
• changeEventsProcessed: 데이터베이스 스트림에서 처리된 변경 이벤트 수예요.
• changeEventsProcessingErrors: 데이터베이스 스트림의 변경 이벤트 처리 오류 수예요.
• bytesReceived: 소스가 수신한 바이트의 총 개수예요.
• bytesProcessed: 소스가 처리한 바이트의 총 개수예요.
• positiveAcknowledgementSets: 스트림 처리에서 긍정 승인된 승인 세트 수예요.
• negativeAcknowledgementSets: 스트림 처리에서 부정 승인된 승인 세트 수예요.
• checkpointCount: 스트림 처리에서 체크포인트의 총 개수예요.
• noDataExtendLeaseCount: 마지막 체크포인트 이후 처리된 새 데이터가 없는 파티션에서 리스가 연장된 횟수예요.
• giveupPartitionCount: 파티션이 포기된 횟수예요.
• replicationLogEntryProcessingTime: 복제 로그 이벤트를 처리하는 데 걸린 시간이에요.
• replicationLogEntryProcessingErrors: 처리에 실패한 복제 로그 이벤트 수예요.

더 알아보기 (Learn more)

• 원문 전체: https://docs.opensearch.org/latest/data-prepper/pipelines/configuration/sources/rds/