Redshift最佳实践之Aurora/DMS数据同步、无服务器架构以及权限管理

随着数据价值的突显，越来越多的客户期望建立准实时数据分析系统来为业务提供数据支撑，因此亟需建立一个数据仓库并将业务数据近乎实时地同步到数据仓库中进行查询和统计。本博文以Redshift作为数仓，并使用 AWS Database Migration Service (AWS DMS) 及其更改数据捕获 (CDC) 功能将Aurora 数据库的数据同步到Redshift的方案来阐述系统实施过程中遇到痛点及解决办法。

业务描述：

Redshift主要承担下述负载
1，上游数据写入及供内部下游业务系统查询
2，承担ETL任务，1分钟短时任务占15%，5分钟任务占19%，15分钟任务占66%
3，为webportal提供查询数据支撑
4，为BI 工具查询分析提供数据支撑

痛点：

1，当遇到Redshift维护窗口时Aurora通过DMS同步数据至Redshift任务异常

2，当遇到大批量ETL触发时预置Redshift资源紧张影响查询统计性能

3，基于列级别权限访问控制

1，同步数据从Aurora到Redshift

痛点：使用DMS从Aurora同步数据到Redshift任务异常

名词解释

CDC：CDC的全称是Change Data Capture（变更数据获取），其核心思想是监测并捕获数据库的变动（包括数据表的插入、更新和删除），将变动数据按照发生顺序完整记录并输送到目标数据库中。

DMS：AWS Database Migration Service (AWS DMS) 是一项云服务，可轻松迁移关系数据库、数据仓库、NoSQL 数据库及其他类型的数据存储。

挑战

DMS复制任务进行CDC追加写入数据到Redshift时，不可避免地会遇到因Redshift维护窗口期服务不可用的问题，此外Redshift维护时段每周发生一次且不可关闭。在Redshift维护窗口期间，DMS的CDC任务持续进行就会潜在导致DMS任务异常及数据丢失。

解决方案：

1，将Redshift维护事件发送到SNS ，步骤见： https://docs.aws.amazon.com/redshift/latest/mgmt/manage-event-notifications-console.html

2，通过Lambda监听SNS事件停止/启动DMS任务

import json
import json
import boto3
import time
from datetime import datetime
## Main body of the code
dms_client = boto3.client('dms')
##源端ARN
endpoint_arn = 'arn:aws:dms:<region>:<account_no>:endpoint:<endpoint_id>'

    
def stop_all_dms_task():
    response = dms_client.describe_replication_tasks(
        Filters=[
            {
                'Name': 'endpoint-arn',
                'Values': [endpoint_arn]
            },
        ],
        MaxRecords=100,
        Marker='string',
        WithoutSettings=True
    )
    response = [dms_client.stop_replication_task(ReplicationTaskArn=item['ReplicationTaskArn']) \
                for item in response['ReplicationTasks'] if item['Status'] in ['starting','running']]
    print('停止{}个任务完成.'.format(len(response)))
    return response
def start_dms_task(task_arn):
    try:
        dms_client.start_replication_task(ReplicationTaskArn=task_arn,StartReplicationTaskType='resume-processing')
    except Exception as e:
        print('任务：{},启动异常：{}'.format(task_arn,str(e)))
def start_all_dms_task():
    print('开始启动任务...')
    response = dms_client.describe_replication_tasks(
        Filters=[
            {
                'Name': 'endpoint-arn',
                'Values': [endpoint_arn]
            },
        ],
        MaxRecords=100,
        Marker='string',
        WithoutSettings=True
    )
    # print(response)
    res = [start_dms_task(item['ReplicationTaskArn']) \
                for item in response['ReplicationTasks'] if item['Status'] in ['stopped']]
    # print('启动结果：{}'.format(res))
    stopping_task = [item['ReplicationTaskArn'] \
                for item in response['ReplicationTasks'] if item['Status'] in ['stopping']]
    # print(stopping_task)
    if stopping_task:
        print('启动{}个任务完成,等待1min拉起正在停止的{}个任务.'.format(len(res), len(stopping_task)))
        time.sleep(60)
        start_all_dms_task()
    else:
        print('所有任务全部启动完成.')
# Add the event id's you want to alert on from this link to the list below
# https://docs.aws.amazon.com/redshift/latest/mgmt/working-with-event-notifications.html
# check_list = ['2003','2004','3519','3520']
redshift_event_trigger = {
    '2003': stop_all_dms_task,
    '2004': start_all_dms_task,
    '3519': None,
    '3520': None
}

def lambda_handler(event, context):
    message = event['Records'][0]['Sns']['Message']
    print(message)
    message = json.loads(message)
    event_ts = message['Event Time']
    event_msg = message['Event Message']
    print('Event Time : {}, Event Message : {}'.format(event_ts, event_msg))
    eventid = message['About this Event'].split('-')[-1].strip()
    print('eventid : {}'.format(eventid))
    # 事件处理
    handler = redshift_event_trigger.get(eventid)
    if handler:
        handler()

2，数据共享Serverless以提升性能

痛点：当遇到大批量ETL触发时预置Redshift资源紧张影响查询统计性能

名词解释：

1，Redshift预置集群：细粒度控制与定制化集群，需预先根据资源量评估实例类型及集群容量，按实例使用时长固定收费。

2，Redshift Serverless：自动扩展资源，而无需管理数据仓库集群, 具备高并发性能特性，根据标准化计算单元 – Redshift Processing Unit (RPU) 按需付费。

挑战：

当前Redshift预置集群主要进行报表分析、查询统计、ETL等任务。当大批量ETL任务触发时会有排队等待现象，CPU会有短时突增接近100%的峰值，查询响应非常缓慢，一次大的查询接近15分钟，业务体验非常不好。如何在不大幅增加成本的情况下提升查询分析响应速度成为急需解决的问题。

解决方案：
经过分析，我们发现原有的Redshift预置集群在处理DMS CDC任务以及少量ETL 任务的时候，系统负载在恰当的水平（CPU利用率在50%左右），系统负载压力来自于与此同时的查询统计请求及高峰期大量ETL任务。基于此考虑如下几种方案：

方案一、对预置集群进行按需扩缩容来解决资源问题，但面临着成本压力及扩缩容会有短时读写服务中断影响

方案二、配置预置集群进行读写分离，但非高峰期存在资源浪费的情况。

方案三、保留原有的预置集群，新增Redshift serverless共享预置集群数据来实现分担查询分析任务。

经论证，方案三采用新推出的Redshift Serverless无服务器与预置Redshift通过数据共享的方式来分担读查询分析工作可在不大幅增加成本的情况下实现无中断平滑分担负载，充分利用Redshift-Serverless完美动态扩缩容能力来承担查询分析任务，避免空闲期资源浪费。

在新的架构下，Redshift预置集群负责DMS CDC数据摄入任务和Airflow ETL的任务，Redshift无服务器集群负责来自于客户端的查询统计分析工作负载。新的架构维持原有的Redshift预置集群配置不变，在最小变动的情况下将数据查询的负载压力转移到一种更具弹性的无服务器集群。

该解决方案实施包括以下步骤：

1，建立预置Redshift集群，且开启数据加密，并记录命令空间 ID：类似：“xxxxxx-xxxx-xxxx-xxxx-xxxxxxxx”

记录如下图命名空间 ID

2，对于共享数据给Serverless，必须对预置Redshift集群和Serverless集群进行加密，默认Serverless启用加密且不可关闭。，确保预置集群已开启数据加密，如在创建时未启用加密可在后续进行启用

3，建立Redshift Serverless工作组和命名空间，并记录命名空间ID： “aaaaa-aaaa-aaaa-aaaa-aaaaaa”

4，建立数据共享，预置集群端操作

a，建立datashare：CREATE DATASHARE “DataShare-name”;

CREATE DATASHARE DataShareToServerless;

b，添加对象到新建的 DataShare：ALTER DATASHARE “DataShare-name” ADD SCHEMA “Schema-Name”;

ALTER DATASHARE DataShareToServerless ADD SCHEMA PUBLIC;

c，添加Schema下面表到DataShare，可指定表或添加全部：ALTER DATASHARE “DataShare-name” ADD ALL TABLES IN SCHEMA “Schema-Name”;

ALTER DATASHARE DataShareToServerless ADD ALL TABLES IN SCHEMA PUBLIC;

d，授权给对应DB用户

GRANT ALTER, SHARE ON DATASHARE DataShareToServerless TO dbuser;

如需采用IAM用户进行federated user授权，则可在Redshift通过select语句查询 IAM用户在DB里面映射以IAM开头用户进行授权即可

e，授权新建DataSahre给Severless：GRANT USAGE ON DATASHARE “DataShare-name” TO NAMESPACE ‘消费端集群命名空间-ID’;

GRANT USAGE ON DATASHARE DataShareToServerless TO NAMESPACE '8fab2610-f5ac-4416-9dfb-d5c78d5ca8a3';

f，检查Datashare状态

DESC DATASHARE DataShareToServerless; 如共享成本则可看到如下共享信息，至此预置集群端操作完成

5，Serverless端配置以验证预置集群共享是否成功及访问读取数据

A，检查Datashare状态和信息, 执行SHOW DATASHARES; 应看到预置集群所共享Datashare信息如下

查看对应Datashare详细信息：DESC DATASHARE DataShareToServerless OF NAMESPACE '524ee257-3dfc-4936-9da4-8ed1fa467830';

B，在Serverless集群端通过Datashare建立DB：

CREATE DATABASE reader_db FROM DATASHARE DataShareToServerless OF NAMESPACE '524ee257-3dfc-4936-9da4-8ed1fa467830';

C，建立外部Schema实现跨数据库查询，schema名字可按业务命名规范进行设计以避免业务迁移过程中改动过多代码

CREATE EXTERNAL SCHEMA reader_schema FROM REDSHIFT DATABASE 'reader_db' SCHEMA 'public';

D，验证访问数据，从Serverless端可以看到预置Redshift集群所共享数据，且能进行数据访问。

至此，预置Redshift集群共享数据给Serverless配置完成，后续查询分析类任务可以在Serverless侧进行，无需担心扩缩容及空闲期资源浪费的情况。

3，列级别细粒度访问控制

痛点：基于列级别实现细粒度权限访问控制

名词解释：
角色：角色是可以分配给用户或其他角色的权限集合。
RBAC: 借助基于角色的访问控制 (RBAC, Role-Based Access Control) 在 Amazon Redshift 中管理数据库权限，可以简化 Amazon Redshift 中的安全权限管理

挑战：
安全部门要求对共享的数据做到列级别的细粒度访问控制。由于用户众多且其权限存在变化的可能，因此安全部分希望以一种最小的维护方式对用户进行列级别访问控制。

解决方案：
通过 RBAC 将列权限分配给角色，安全部门可以将角色分配给用户。为了保证每个表都是列级权限控制的，我们使用 ALTER DEFAULT PRIVILEGES 来定义默认访问权限集，这些默认访问权限集可以应用于未来创建的表。在创建完表之后，我们可以通过REVOKE 语句回收所有列的访问权限，再为每个角色创建列的权限，达到列级别权限访问控制的目的。

例子：

1，创建角色

create role analyzer_full_previleges;
create role analyzer_limit_previleges;

2，默认回收所有角色的权限

alter default privileges in schema public revoke all on tables from role analyzer_full_previleges;
alter default privileges in schema public revoke all on tables from role analyzer_limit_previleges;

3，创建用户，将角色赋予用户

create user analyzer_full_select with password '@AbC4321!';
grant role analyzer_full_previleges to analyzer_full_select;

create user analyzer_limit_select with password '@AbC4321!';
grant role analyzer_limit_previleges to analyzer_limit_select;

4，创建表

create table public.table_1 (
col1 varchar(20),
col2 varchar(20),
col3 varchar(20),
col4 varchar(20),
col5 varchar(20)
);

insert into public.table_1 values ('col1_value', 'col2_value', 'col3_value', 'col4_value', 'col5_value');

5，给角色赋予列权限

为角色赋予表的SELECT权限

grant SELECT (col1, col2, col3, col4, col5) ON TABLE public.table_1 to ROLE analyzer_full_previleges;
grant SELECT (col1, col2) ON TABLE public.table_1 to ROLE analyzer_limit_previleges;

6，测试列级别权限访问控制的效果

A，用户analyzer_full_select可以查询所有的列

set session authorization 'analyzer_full_select';
select * from public.table_1;

B，用户analyzer_limit_select只能查询col1, col2

set session authorization 'analyzer_limit_select';
select col1, col2 from public.table_1;

否则，系统提示权限相关错误

set session authorization 'analyzer_limit_select';
select col1, col2, col3 from public.table_1;

结束语

随着数据价值的突显，未来数据分析将会更加重要，Redshift已具备预置集群与Severless无缝集成，自动升级，无宕机运维，安全加密，自动扩展，计算资源按需分配等能力，相信未来围绕Redshift的工具生态链会更加丰富，Redshift自身产品功能也会更加完善，成为数据分析的旗舰产品。

参考文档：

https://docs.aws.amazon.com/zh_cn/redshift/latest/dg/within-account.html

https://docs.aws.amazon.com/zh_cn/dms/latest/userguide/Welcome.html

https://docs.aws.amazon.com/redshift/latest/dg/t_Roles.html

https://docs.aws.amazon.com/redshift/latest/dg/r_roles-alter-default-privileges.html

亚马逊AWS官方博客

Redshift最佳实践之Aurora/DMS数据同步、无服务器架构以及权限管理

业务描述：

痛点：

1，同步数据从Aurora到Redshift

2，数据共享Serverless以提升性能

3，列级别细粒度访问控制

结束语

本篇作者