ApsaraDB RDS:Instance-level migration of SQL Server to the cloud

Prepare the OSS bucket

If a bucket already exists, verify that it meets these requirements:

• Storage class: Standard. Infrequent Access, Archive, Cold Archive, and Deep Cold Archive are not supported.

• Server-side encryption: Disabled.

If no bucket exists, activate OSS and then create one:

1. Log on to the OSS console, click Buckets, and then click Create Bucket.

2. Configure these parameters. Leave all others at their default values.

   Important

   - Create this bucket specifically for the migration. Delete it after migration is complete to prevent data exposure and reduce costs. - Do not enable server-side encryption when creating the bucket.

   Parameter	Description	Example
   Bucket name	Globally unique name that cannot be changed after creation. Use only lowercase letters, digits, and hyphens (-); start and end with a letter or digit; 3–63 characters.	migratetest
   Region	Must match the region of your ApsaraDB RDS instance.	China (Hangzhou)
   Storage class	Select Standard.	Standard

Upload the backup files

Choose the upload method based on file size and your operating system:

Use ossbrowser (recommended for most cases)

1. Download ossbrowser.

2. Extract the downloaded package and launch oss-browser.exe (Windows x64 example).

3. Select Log On With AK, enter your AccessKey ID and AccessKey secret, keep the default values for other fields, and click Log On.

   Note

   An AccessKey pair is used for identity verification. Keep it confidential.

   

4. Click the destination bucket.

   

5. Click , select the backup files, and click Open.

Use the OSS console (files smaller than 5 GB)

1. Log on to the OSS console.

2. Click Buckets, then click the name of the destination bucket.

   

3. In the Objects list, click Upload Object.

   

4. Drag backup files to the Files to Upload area, or click Select Files.

   

5. Click Upload Object at the bottom of the page.

Use the OSS API for multipart upload (files larger than 5 GB)

Set the following environment variables before running the script:

# -*- coding: utf-8 -*-
"""
Alibaba Cloud OSS Python SDK v2
Dependency: pip install alibabacloud-oss-v2
"""

import os
import sys
from pathlib import Path
import alibabacloud_oss_v2 as oss
from alibabacloud_oss_v2 import exceptions as oss_ex


def get_client_from_env(region: str, endpoint: str | None = None) -> oss.Client:
    """
    Create a v2 client from environment variables.
    - Prioritize using Region (recommended), but also support custom Endpoints (optional).
    - Compatible with both AK and STS:
        * AK: Requires OSS_ACCESS_KEY_ID / OSS_ACCESS_KEY_SECRET
        * STS: Also requires OSS_SESSION_TOKEN (compatible with the old variable OSS_SECURITY_TOKEN)
    """
    # Compatibility: If the user uses the old variable OSS_SECURITY_TOKEN, map it to the v2 expected OSS_SESSION_TOKEN
    sec_token_legacy = os.getenv("OSS_SECURITY_TOKEN")
    if sec_token_legacy and not os.getenv("OSS_SESSION_TOKEN"):
        os.environ["OSS_SESSION_TOKEN"] = sec_token_legacy

    ak = os.getenv("OSS_ACCESS_KEY_ID")
    sk = os.getenv("OSS_ACCESS_KEY_SECRET")
    st = os.getenv("OSS_SESSION_TOKEN")  # STS Token (optional)

    if not (ak and sk):
        raise ValueError("No valid AK found. Set the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables. "
                         "If using STS, also set OSS_SESSION_TOKEN (or the old name OSS_SECURITY_TOKEN).")

    # Indicate the type of credential used
    if st:
        print("STS Token (OSS_SESSION_TOKEN) detected. Using STS credentials.")
    else:
        print("No STS Token detected. Using AccessKey (AK) credentials.")

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Basic network configuration
    cfg.region = region  # Example: 'cn-hangzhou'
    if endpoint:
        # Optional: Custom Endpoint (e.g., internal network, accelerated, dedicated domain)
        cfg.endpoint = endpoint

    # You can also add other configurations here, such as: cfg.use_accelerate_endpoint = True
    return oss.Client(cfg)


def resumable_upload_file_v2(
    client: oss.Client,
    bucket_name: str,
    object_key: str,
    file_path: str,
    part_size: int = 1 * 1024 * 1024,
    parallel_num: int = 4,
    checkpoint_dir: str | None = None,
):
    """
    Implement concurrent multipart upload with resumable upload.

    :param client: Initialized oss.Client
    :param bucket_name: Destination bucket name
    :param object_key: Destination object key (without bucket name)
    :param file_path: Full path of the local file
    :param part_size: Part size in bytes, default is 1 MB
    :param parallel_num: Number of concurrent upload threads, default is 4
    :param checkpoint_dir: Directory to store breakpoint information; if None, resumable upload is disabled
    """
    file_path = str(file_path)
    if not Path(file_path).exists():
        raise FileNotFoundError(f"Error: Local file not found. Check the file_path configuration: {file_path}")

    # Construct the Uploader; enable resumable upload based on whether checkpoint_dir is provided
    if checkpoint_dir:
        uploader = client.uploader(
            enable_checkpoint=True,
            checkpoint_dir=checkpoint_dir,
            part_size=part_size,
            parallel_num=parallel_num,
        )
    else:
        uploader = client.uploader(
            part_size=part_size,
            parallel_num=parallel_num,
        )

    print(f"Starting to upload file: {file_path}")
    print(f"Destination Bucket: {bucket_name}")
    print(f"Destination Object: {object_key}")
    print(f"Part size: {part_size} bytes, Concurrency: {parallel_num}")
    if checkpoint_dir:
        print(f"Resumable upload: Enabled (checkpoint_dir={checkpoint_dir})")
    else:
        print("Resumable upload: Disabled (set checkpoint_dir to enable)")

    # Execute the upload (Uploader automatically chooses between multi/single part concurrent upload based on size)
    result = uploader.upload_file(
        oss.PutObjectRequest(bucket=bucket_name, key=object_key),
        filepath=file_path,
    )

    print("-" * 30)
    print("File uploaded successfully!")
    print(f"HTTP Status: {result.status_code}")
    print(f"ETag: {result.etag}")
    print(f"Request ID: {result.request_id}")
    # CRC-64 checksum; v2 enables data validation by default
    print(f"CRC64: {result.hash_crc64}")
    print("-" * 30)


def main():
    # Before running the code example, make sure you have set the corresponding environment variables.
    # macOS/Linux:
    #   AK method:
    #     export OSS_ACCESS_KEY_ID=YOUR_AK_ID
    #     export OSS_ACCESS_KEY_SECRET=YOUR_AK_SECRET
    #   STS method:
    #     export OSS_ACCESS_KEY_ID=YOUR_STS_ID
    #     export OSS_ACCESS_KEY_SECRET=YOUR_STS_SECRET
    #     export OSS_SECURITY_TOKEN=YOUR_STS_TOKEN
    #
    # Windows:
    #   Powershell: $env:OSS_ACCESS_KEY_ID="YOUR_AK_ID"
    #   cmd: set OSS_ACCESS_KEY_ID=YOUR_AK_ID

    # ===================== Parameters (modify as needed) =====================
    # Region example: 'cn-hangzhou'; use Region first
    region = "cn-hangzhou"

    # Optional: Custom Endpoint (for internal network, dedicated domain, accelerated domain name, etc.)
    # Example: 'https://oss-cn-hangzhou.aliyuncs.com'
    endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'

    # Bucket and Object
    bucket_name = "examplebucket"
    object_key = "test.bak"

    # Full path of the local file to upload.
    # Windows example: r'D:\localpath\examplefile.txt'  (note the r at the beginning)
    # macOS/Linux example: '/Users/test/examplefile.txt'
    file_path = r"D:\oss\test.bak"

    # Sharding and concurrency
    part_size = 1 * 1024 * 1024  # Default is 1 MB; OSS requires a minimum part size of 100 KB
    parallel_num = 4

    # Resumable upload directory (pass None to disable; specify a writable directory)
    checkpoint_dir = str(Path.cwd() / ".oss_checkpoints")
    # =================== End of parameters ===================

    print("Script execution starts...")
    try:
        client = get_client_from_env(region=region, endpoint=endpoint)
        # If resumable upload is enabled, make sure the directory exists
        if checkpoint_dir:
            Path(checkpoint_dir).mkdir(parents=True, exist_ok=True)

        resumable_upload_file_v2(
            client=client,
            bucket_name=bucket_name,
            object_key=object_key,
            file_path=file_path,
            part_size=part_size,
            parallel_num=parallel_num,
            checkpoint_dir=checkpoint_dir,
        )
    except FileNotFoundError as e:
        print(e)
    except oss_ex.ServiceError as e:
        # Error returned by the OSS server
        print("\nAn OSS server-side error occurred.")
        print(f"HTTP Status: {getattr(e, 'status_code', 'N/A')}")
        print(f"Error Code: {getattr(e, 'code', 'N/A')}")
        print(f"Message: {getattr(e, 'message', 'N/A')}")
        print(f"Request ID: {getattr(e, 'request_id', 'N/A')}")
        print(f"Endpoint: {getattr(e, 'request_target', 'N/A')}")
    except oss_ex.BaseError as e:
        # SDK local/serialization/deserialization/credential errors
        print("\nAn OSS SDK client-side error occurred.")
        print(str(e))
    except Exception as e:
        print(f"\nAn unknown error occurred: {e}")


if __name__ == "__main__":
    main()

Script and API errors

OpenAPI error codes