Alibaba Cloud DTS 증분 데이터 마이그레이션 모범 사례 (Function)¶

전제 조건¶

Datakit 설치¶

[통합] 모듈, [Datakit]를 클릭하고, 운영 체제 및 시스템 유형에 따라 적절한 설치 명령을 선택합니다.

Datakit 설치 명령을 복사하여 모니터링 대상 서버에서 직접 실행합니다.
설치 디렉터리 /usr/local/datakit/
로그 디렉터리 /var/log/datakit/
기본 구성 파일 /usr/local/datakit/conf.d/datakit.conf
플러그인 구성 디렉터리 /usr/local/datakit/conf.d/

Function 설치¶

[통합] 모듈, [Function]을 클릭하고, 설치 패키지를 다운로드한 후 설치 명령을 실행합니다.

설치가 완료되면 브라우저를 사용하여 http://서버IP주소:8088에 접속하여 초기화 작업 화면을 진행합니다.

기본 사용자 이름/비밀번호 admin으로 시스템에 로그인합니다.

RAM 액세스 제어¶

RAM 콘솔 https://ram.console.aliyun.com/users에 로그인합니다.
사용자 생성: 인사 관리 - 사용자 - 사용자 생성

AccessKey ID 및 AccessKey Secret의 CSV 파일을 저장하거나 다운로드합니다. (구성 파일에서 사용됨)
사용자 권한 부여 (데이터 전송 서비스(DTS)에 대한 읽기 전용 액세스 권한)

스크립트 개발¶

Alibaba Cloud DTS 증분 데이터 마이그레이션¶

Alibaba Cloud DTS 제품 문서 - 신규 API - DTS 작업 세부 정보 조회에 접속하여 디버그 모드로 진입합니다.
SDK 종속성 정보 - 신규 SDK를 찾아 SDK 설치 명령을 복사합니다.

관리 - 실험적 기능 - PIP 도구 모듈 활성화

Alibaba Cloud SDK 종속 패키지 설치

새 스크립트 세트를 만들고 스크립트를 추가합니다.

코드를 작성합니다. AccessKey ID, AccessKey Secret, Region을 입력해야 합니다.

참고 문서:

# -*- coding: utf-8 -*-

import sys
import json
from typing import List

from alibabacloud_dts20200101.client import Client as Dts20200101Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dts20200101 import models as dts_20200101_models

# 阿里云 sdk
class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(
        access_key_id: str,
        access_key_secret: str,
    ) -> Dts20200101Client:
        """
        使用AK&SK初始化账号Client
        @param access_key_id:
        @param access_key_secret:
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # 您的AccessKey ID,
            access_key_id=access_key_id,
            # 您的AccessKey Secret,
            access_key_secret=access_key_secret
        )
        # 访问的域名
        config.endpoint = 'dts.cn-hangzhou.aliyuncs.com'
        return Dts20200101Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client('accessKeyId', 'accessKeySecret')
        describe_dts_jobs_request = dts_20200101_models.DescribeDtsJobsRequest(region='cn-hangzhou')
        client.describe_dts_jobs(describe_dts_jobs_request)

@DFF.API('dts_demo', timeout=300)
def dts_demo():
    # 实例化sample
    sample = Sample()
    client = sample.create_client(
        access_key_id="AccessKey ID",
        access_key_secret="AccessKey Secret"
    )
    describe_dts_jobs_request = dts_20200101_models.DescribeDtsJobsRequest(
    # DTS实例的任务类型。MIGRATION；SYNC；SUBSCRIBE。
            region='region',job_type='MIGRATION'
        )
    response = client.describe_dts_jobs(describe_dts_jobs_request).to_map()
    dtsjoblist = response['body']['DtsJobList'][0]
    dts_instanceid = dtsjoblist['DtsInstanceID']
    dts_jobname = dtsjoblist['DtsJobName']
    # 增量数据迁移或同步的状态指标
    datasynchronization = response['body']['DtsJobList'][0]['DataSynchronizationStatus']
    dts_status = datasynchronization['Status']
    dts_percent = datasynchronization['Percent']

    # 行协议数据
    points = [
        {'measurement': 'aliyun_api_dts',
         'tags': {'instanceId': dts_instanceid,'jobname': dts_jobname}, 
         'fields': {'dts_status': dts_status,'dts_percent':dts_percent}
        }
    ]
    write_metrics(points)

def write_metrics(points: list):
    datakit = DFF.SRC('datakit')
    res = datakit.write_metric_many(points)