把 RDS MySQL 数据入仓 Amazon Redshift 的三种方式 · 首推 Zero-ETL

aws s3 cp user_test.tsv s3://<your-bucket>/ingest/user_test.tsv --region us-east-1
aws s3 ls s3://<your-bucket>/ingest/

DROP TABLE IF EXISTS public.user_test;
CREATE TABLE public.user_test (
    id         BIGINT,
    name       VARCHAR(50),
    email      VARCHAR(100),
    age        SMALLINT,
    created_at TIMESTAMP
)
DISTSTYLE AUTO
SORTKEY (created_at);

COPY public.user_test (id, name, email, age, created_at)
FROM 's3://<your-bucket>/ingest/user_test.tsv'
IAM_ROLE '<redshift-copy-role-arn>'
DELIMITER '\t'
IGNOREHEADER 1
REGION 'us-east-1';

-- 若是标准 CSV，把 DELIMITER 换成 CSV：
-- CSV IGNOREHEADER 1

SELECT COUNT(*) FROM public.user_test;          -- 应等于源表行数
SELECT * FROM public.user_test ORDER BY id LIMIT 5;

-- 加载失败时看逐行错误原因
SELECT * FROM sys_load_error_detail
ORDER BY start_time DESC LIMIT 10;

CREATE TABLE IF NOT EXISTS public.address_glue (
    id         INTEGER NOT NULL,
    street     VARCHAR(255),
    city       VARCHAR(100),
    state      VARCHAR(100),
    zip_code   VARCHAR(20),
    country    VARCHAR(100),
    created_at TIMESTAMP,
    PRIMARY KEY (id)
) DISTSTYLE AUTO SORTKEY (id);

aws glue create-connection --region us-east-1 --connection-input '{
  "Name": "glue-mysql-conn",
  "ConnectionType": "JDBC",
  "ConnectionProperties": {
    "JDBC_CONNECTION_URL": "jdbc:mysql://<rds-endpoint>:3306/<db>",
    "JDBC_ENFORCE_SSL": "true",
    "SECRET_ID": "<mysql-secret-arn>"
  },
  "PhysicalConnectionRequirements": {
    "SubnetId": "<private-subnet-id>",
    "SecurityGroupIdList": ["<glue-sg-id>"],
    "AvailabilityZone": "us-east-1b"
  }
}'

aws glue create-connection --region us-east-1 --connection-input '{
  "Name": "glue-redshift-native-conn",
  "ConnectionType": "REDSHIFT",
  "ConnectionProperties": {
    "HOST": "<workgroup>.<account>.us-east-1.redshift-serverless.amazonaws.com",
    "PORT": "5439",
    "DATABASE": "dev",
    "ROLE_ARN": "<glue-job-role-arn>"
  },
  "AuthenticationConfiguration": {
    "AuthenticationType": "BASIC",
    "SecretArn": "<redshift-secret-arn>"
  },
  "PhysicalConnectionRequirements": {
    "SubnetId": "<private-subnet-id>",
    "SecurityGroupIdList": ["<glue-sg-id>"],
    "AvailabilityZone": "us-east-1b"
  },
  "ValidateForComputeEnvironments": ["SPARK"]
}'
# 创建返回 IN_PROGRESS，几十秒后应变 READY，READY 才能用
aws glue get-connection --region us-east-1 \
  --name glue-redshift-native-conn --query 'Connection.Status'

aws glue create-database --region us-east-1 \
  --database-input '{"Name":"mysql_src"}'
aws glue create-crawler --region us-east-1 \
  --name src-mysql-crawler --role <glue-job-role> --database-name mysql_src \
  --targets '{"JdbcTargets":[{"ConnectionName":"glue-mysql-conn","Path":"<db>/%"}]}'
aws glue start-crawler --region us-east-1 --name src-mysql-crawler

import sys
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc = SparkContext()
glueContext = GlueContext(sc)
job = Job(glueContext)
job.init(args["JOB_NAME"], args)
# 读 MySQL 源
src = glueContext.create_dynamic_frame.from_options(
    connection_type="mysql",
    connection_options={
        "useConnectionProperties": "true",
        "connectionName": "glue-mysql-conn",
        "dbtable": "address",
    },
    transformation_ctx="src",          # 开书签必须有 ctx
)
# 写 Redshift 目标
glueContext.write_dynamic_frame.from_options(
    frame=src,
    connection_type="redshift",
    connection_options={
        "useConnectionProperties": "true",
        "connectionName": "glue-redshift-native-conn",
        "database": "dev",
        "dbtable": "public.address_glue",
        "redshiftTmpDir": "s3://<your-bucket>/temporary/",
    },
    transformation_ctx="dst",
)
job.commit()
aws s3 cp mysql-to-redshift.py s3://<your-bucket>/scripts/mysql-to-redshift.py

aws glue create-job --region us-east-1 \
  --name mysql-to-redshift-sync \
  --role <glue-job-role-arn> \
  --command '{"Name":"glueetl","ScriptLocation":"s3://<your-bucket>/scripts/mysql-to-redshift.py","PythonVersion":"3"}' \
  --connections '{"Connections":["glue-mysql-conn","glue-redshift-native-conn"]}' \
  --default-arguments '{
      "--job-language":"python",
      "--TempDir":"s3://<your-bucket>/temporary/",
      "--enable-metrics":"true",
      "--enable-continuous-cloudwatch-log":"true",
      "--job-bookmark-option":"job-bookmark-disable"
  }' \
  --glue-version "4.0" \
  --worker-type G.1X --number-of-workers 2 \
  --max-retries 0 --timeout 20

RUN_ID=$(aws glue start-job-run --region us-east-1 \
          --job-name mysql-to-redshift-sync --query JobRunId --output text)
# 期望 RUNNING → SUCCEEDED；失败时 ErrorMessage 里有根因
aws glue get-job-run --region us-east-1 \
  --job-name mysql-to-redshift-sync --run-id "$RUN_ID" \
  --query 'JobRun.[JobRunState,ErrorMessage]'

SELECT t.table_name
FROM information_schema.tables t
LEFT JOIN information_schema.table_constraints c
       ON c.table_schema = t.table_schema
      AND c.table_name   = t.table_name
      AND c.constraint_type = 'PRIMARY KEY'
WHERE t.table_schema = '<db>'
  AND t.table_type   = 'BASE TABLE'
  AND c.constraint_name IS NULL;   -- 输出的表缺主键，需补

CREATE DATABASE zeroetl_db FROM INTEGRATION '<集成ID>';
-- 集成 ID = 集成 ARN 的最后一段

-- permission denied to create database  → 换 Admin 用户
-- Unknown arn ... pair                  → 删掉多写的 DATABASE 子句

-- 初始加载：应与源库行数一致
SELECT COUNT(*) FROM zeroetl_db.<db>.address;
SELECT * FROM zeroetl_db.<db>.address ORDER BY id LIMIT 5;

-- 看每张表的同步状态（哪张没同步、为什么）
SELECT * FROM SVV_INTEGRATION_TABLE_STATE;