Conceitos básicos das integrações ETL zero do Aurora com o Amazon Redshift
Antes de criar uma integração ETL zero com o Amazon Redshift, configure o cluster de banco de dados do Aurora e o data warehouse do Amazon Redshift com os parâmetros e as permissões necessários. Durante a configuração, você realizará as seguintes etapas:
Depois de concluir essas tarefas, prossiga para Criar integrações ETL zero do Amazon Aurora com o Amazon Redshift.
É possível usar os SDKs da AWS para automatizar o processo de configuração. Para ter mais informações, consulte Configurar uma integração usando SDKs da AWS.
Etapa 1: Criar um grupo de parâmetros de cluster de banco de dados personalizado
As integrações ETL zero do Aurora com o Amazon Redshift exigem valores específicos para os parâmetros de cluster de banco de dados do Aurora que controlam a replicação. Especificamente, o Aurora MySQL requer o log binário aprimorado (aurora_enhanced_binlog) e o Aurora PostgreSQL requer replicação lógica aprimorada (aurora.enhanced_logical_replication).
Para configurar o registro em log binário ou a replicação lógica, primeiro é necessário criar um grupo de parâmetros de cluster de banco de dados personalizado e, depois, associá-lo ao cluster de banco de dados de origem.
Aurora MySQL (família aurora-mysql8.0):
Além disso, verifique se o parâmetro binlog_transaction_compression não está definido como ON e se o parâmetrobinlog_row_value_options não está definido como PARTIAL_JSON.
Para ter mais informações sobre o log binário aprimorado do Aurora MySQL, consulte Configurar o log binário avançado para Aurora MySQL.
Aurora PostgreSQL (família aurora-postgresql16):
-
rds.logical_replication=1
-
aurora.enhanced_logical_replication=1
-
aurora.logical_replication_backup=0
-
aurora.logical_replication_globaldb=0
Habilitar a replicação lógica aprimorada (aurora.enhanced_logical_replication) sempre gravará todos os valores de coluna no log de gravação antecipada (WAL), mesmo que REPLICA IDENTITY FULL não esteja habilitada. Isso pode aumentar as IOPS do cluster de banco de dados de origem.
Se você desabilitar o parâmetro de cluster de banco de dados aurora.enhanced_logical_replication após a criação da integração, a instância de banco de dados primária invalidará todos os slots de replicação lógica. Isso interrompe a replicação da origem no destino, e você deve recriar os slots de replicação na instância de banco de dados primária. Para evitar interrupções, mantenha o parâmetro habilitado de maneira consistente durante a replicação.
Etapa 2: Selecionar ou criar um cluster de banco de dados de origem
Depois de criar um grupo de parâmetros de cluster de banco de dados personalizado, selecione ou crie um cluster de banco de dados do Aurora. Esse cluster será a origem da replicação de dados para o Amazon Redshift. Você pode especificar um cluster de banco de dados que usa instâncias de banco de dados provisionadas ou instâncias de banco de dados do Aurora Serverless v2. Consulte instruções para criar um cluster de banco de dados, consulte Criar um cluster de bancos de dados do Amazon Aurora ou Criar um cluster de banco de dados que usa o Aurora Serverless v2.
O banco de dados deve estar executando uma versão de mecanismo de banco de dados compatível. Para conferir uma lista de versões compatíveis, consulte Regiões e mecanismos de banco de dados do Aurora compatíveis com integrações ETL zero com o Amazon Redshift.
Ao criar o banco de dados, em Configuração adicional, altere o grupo de parâmetros de cluster de banco de dados padrão para o grupo de parâmetros personalizado que você criou na etapa anterior.
Etapa 3: Criar um data warehouse de destino do Amazon Redshift
Depois de criar o cluster de banco de dados, será necessário criar e configurar um data warehouse de destino no Amazon Redshift. O data warehouse deve cumprir os seguintes requisitos:
Para obter instruções sobre como criar um data warehouse, consulte Criar um cluster para clusters provisionados ou Criar um grupo de trabalho com um namespace para o Redshift Serverless.
Ative a distinção entre maiúsculas e minúsculas no data warehouse
Para que a integração seja bem-sucedida, o parâmetro de diferenciação de maiúsculas e minúsculas (enable_case_sensitive_identifier) deve estar ativado para o data warehouse. Por padrão, a distinção entre maiúsculas e minúsculas é desativada em todos os clusters provisionados e grupos de trabalho do Redshift Serverless.
Para ativar a distinção entre maiúsculas e minúsculas, execute as seguintes etapas, dependendo do tipo de data warehouse:
-
Cluster provisionado: para ativar a distinção entre maiúsculas e minúsculas em um cluster provisionado, crie um grupo de parâmetros personalizado com o parâmetro enable_case_sensitive_identifier ativado. Em seguida, associe o grupo de parâmetros ao cluster. Para obter instruções, consulte Gerenciar grupos de parâmetros usando o console ou Configurar valores de parâmetros usando a AWS CLI.
Lembre-se de reinicializar o cluster depois de associar o grupo de parâmetros personalizado a ele.
-
Grupo de trabalho de tecnologia sem servidor: para ativar a distinção entre maiúsculas e minúsculas em um grupo de trabalho do Redshift Serverless, você deve usar a AWS CLI. Atualmente, o console do Amazon Redshift não é compatível com a modificação dos valores dos parâmetros do Redshift Serverless. Envie a seguinte solicitação de atualização do grupo de trabalho:
aws redshift-serverless update-workgroup \
--workgroup-name target-workgroup \
--config-parameters parameterKey=enable_case_sensitive_identifier,parameterValue=true
Não é necessário reinicializar um grupo de trabalho após modificar seus valores de parâmetros.
Configurar a autorização para o data warehouse
Depois de criar um data warehouse, você deve configurar o cluster de banco de dados do Aurora de origem como uma origem de integração autorizada. Para obter instruções, consulte Configurar a autorização para o data warehouse do Amazon Redshift.
Configurar uma integração usando SDKs da AWS
Em vez de configurar cada recurso manualmente, é possível executar o script Python a seguir para configurar automaticamente os recursos necessários. O exemplo de código usa o AWS SDK for Python (Boto3) para criar um cluster de banco de dados do Amazon Aurora de origem e um data warehouse do Amazon Redshift de destino, cada um com os valores de parâmetros necessários. Depois, ele espera que os bancos de dados estejam disponíveis antes de criar uma integração ETL zero entre eles. É possível comentar diferentes funções, dependendo dos recursos que você precisa configurar.
Execute os comandos a seguir para instalar as dependências necessárias:
pip install boto3
pip install time
No script, modifique opcionalmente os nomes dos grupos de parâmetros, a origem e o destino. A função final cria uma integração denominada my-integration após a configuração dos recursos.
- Aurora MySQL
-
import boto3
import time
# Build the client using the default credential configuration.
# You can use the CLI and run 'aws configure' to set access key, secret
# key, and default Region.
rds = boto3.client('rds')
redshift = boto3.client('redshift')
sts = boto3.client('sts')
source_cluster_name = 'my-source-cluster' # A name for the source cluster
source_param_group_name = 'my-source-param-group' # A name for the source parameter group
target_cluster_name = 'my-target-cluster' # A name for the target cluster
target_param_group_name = 'my-target-param-group' # A name for the target parameter group
def create_source_cluster(*args):
"""Creates a source Aurora MySQL DB cluster"""
response = rds.create_db_cluster_parameter_group(
DBClusterParameterGroupName=source_param_group_name,
DBParameterGroupFamily='aurora-mysql8.0',
Description='For Aurora MySQL binary logging'
)
print('Created source parameter group: ' + response['DBClusterParameterGroup']['DBClusterParameterGroupName'])
response = rds.modify_db_cluster_parameter_group(
DBClusterParameterGroupName=source_param_group_name,
Parameters=[
{
'ParameterName': 'aurora_enhanced_binlog',
'ParameterValue': '1',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'binlog_backup',
'ParameterValue': '0',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'binlog_format',
'ParameterValue': 'ROW',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'binlog_replication_globaldb',
'ParameterValue': '0',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'binlog_row_image',
'ParameterValue': 'full',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'binlog_row_metadata',
'ParameterValue': 'full',
'ApplyMethod': 'pending-reboot'
}
]
)
print('Modified source parameter group: ' + response['DBClusterParameterGroupName'])
response = rds.create_db_cluster(
DBClusterIdentifier=source_cluster_name,
DBClusterParameterGroupName=source_param_group_name,
Engine='aurora-mysql',
EngineVersion='8.0.mysql_aurora.3.05.2',
DatabaseName='myauroradb',
MasterUsername='username',
MasterUserPassword='Password01**'
)
print('Creating source cluster: ' + response['DBCluster']['DBClusterIdentifier'])
source_arn = (response['DBCluster']['DBClusterArn'])
create_target_cluster(target_cluster_name, source_arn, target_param_group_name)
response = rds.create_db_instance(
DBInstanceClass='db.r6g.2xlarge',
DBClusterIdentifier=source_cluster_name,
DBInstanceIdentifier=source_cluster_name + '-instance',
Engine='aurora-mysql'
)
return(response)
def create_target_cluster(target_cluster_name, source_arn, target_param_group_name):
"""Creates a target Redshift cluster"""
response = redshift.create_cluster_parameter_group(
ParameterGroupName=target_param_group_name,
ParameterGroupFamily='redshift-1.0',
Description='For Aurora MySQL zero-ETL integrations'
)
print('Created target parameter group: ' + response['ClusterParameterGroup']['ParameterGroupName'])
response = redshift.modify_cluster_parameter_group(
ParameterGroupName=target_param_group_name,
Parameters=[
{
'ParameterName': 'enable_case_sensitive_identifier',
'ParameterValue': 'true'
}
]
)
print('Modified target parameter group: ' + response['ParameterGroupName'])
response = redshift.create_cluster(
ClusterIdentifier=target_cluster_name,
NodeType='ra3.4xlarge',
NumberOfNodes=2,
Encrypted=True,
MasterUsername='username',
MasterUserPassword='Password01**',
ClusterParameterGroupName=target_param_group_name
)
print('Creating target cluster: ' + response['Cluster']['ClusterIdentifier'])
# Retrieve the target cluster ARN
response = redshift.describe_clusters(
ClusterIdentifier=target_cluster_name
)
target_arn = response['Clusters'][0]['ClusterNamespaceArn']
# Retrieve the current user's account ID
response = sts.get_caller_identity()
account_id = response['Account']
# Create a resource policy specifying cluster ARN and account ID
response = redshift.put_resource_policy(
ResourceArn=target_arn,
Policy='''
{
\"Version\":\"2012-10-17\",
\"Statement\":[
{\"Effect\":\"Allow\",
\"Principal\":{
\"Service\":\"redshift.amazonaws.com\"
},
\"Action\":[\"redshift:AuthorizeInboundIntegration\"],
\"Condition\":{
\"StringEquals\":{
\"aws:SourceArn\":\"%s\"}
}
},
{\"Effect\":\"Allow\",
\"Principal\":{
\"AWS\":\"arn:aws:iam::%s:root\"},
\"Action\":\"redshift:CreateInboundIntegration\"}
]
}
''' % (source_arn, account_id)
)
return(response)
def wait_for_cluster_availability(*args):
"""Waits for both clusters to be available"""
print('Waiting for clusters to be available...')
response = rds.describe_db_clusters(
DBClusterIdentifier=source_cluster_name
)
source_status = response['DBClusters'][0]['Status']
source_arn = response['DBClusters'][0]['DBClusterArn']
response = rds.describe_db_instances(
DBInstanceIdentifier=source_cluster_name + '-instance'
)
source_instance_status = response['DBInstances'][0]['DBInstanceStatus']
response = redshift.describe_clusters(
ClusterIdentifier=target_cluster_name
)
target_status = response['Clusters'][0]['ClusterStatus']
target_arn = response['Clusters'][0]['ClusterNamespaceArn']
# Every 60 seconds, check whether the clusters are available.
if source_status != 'available' or target_status != 'available' or source_instance_status != 'available':
time.sleep(60)
response = wait_for_cluster_availability(
source_cluster_name, target_cluster_name)
else:
print('Clusters available. Ready to create zero-ETL integration.')
create_integration(source_arn, target_arn)
return
def create_integration(source_arn, target_arn):
"""Creates a zero-ETL integration using the source and target clusters"""
response = rds.create_integration(
SourceArn=source_arn,
TargetArn=target_arn,
IntegrationName='my-integration'
)
print('Creating integration: ' + response['IntegrationName'])
def main():
"""main function"""
create_source_cluster(source_cluster_name, source_param_group_name)
wait_for_cluster_availability(source_cluster_name, target_cluster_name)
if __name__ == "__main__":
main()
- Aurora PostgreSQL
-
import boto3
import time
# Build the client using the default credential configuration.
# You can use the CLI and run 'aws configure' to set access key, secret
# key, and default Region.
rds = boto3.client('rds')
redshift = boto3.client('redshift')
sts = boto3.client('sts')
source_cluster_name = 'my-source-cluster' # A name for the source cluster
source_param_group_name = 'my-source-param-group' # A name for the source parameter group
target_cluster_name = 'my-target-cluster' # A name for the target cluster
target_param_group_name = 'my-target-param-group' # A name for the target parameter group
def create_source_cluster(*args):
"""Creates a source Aurora PostgreSQL DB cluster"""
response = rds.create_db_cluster_parameter_group(
DBClusterParameterGroupName=source_param_group_name,
DBParameterGroupFamily='aurora-postgresql16',
Description='For Aurora PostgreSQL logical replication'
)
print('Created source parameter group: ' + response['DBClusterParameterGroup']['DBClusterParameterGroupName'])
response = rds.modify_db_cluster_parameter_group(
DBClusterParameterGroupName=source_param_group_name,
Parameters=[
{
'ParameterName': 'rds.logical_replication',
'ParameterValue': '1',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'aurora.enhanced_logical_replication',
'ParameterValue': '1',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'aurora.logical_replication_backup',
'ParameterValue': '0',
'ApplyMethod': 'pending-reboot'
},
{
'ParameterName': 'aurora.logical_replication_globaldb',
'ParameterValue': '0',
'ApplyMethod': 'pending-reboot'
}
]
)
print('Modified source parameter group: ' + response['DBClusterParameterGroupName'])
response = rds.create_db_cluster(
DBClusterIdentifier=source_cluster_name,
DBClusterParameterGroupName=source_param_group_name,
Engine='aurora-postgresql',
EngineVersion='16.4.aurora-postgresql',
DatabaseName='mypostgresdb',
MasterUsername='username',
MasterUserPassword='Password01**'
)
print('Creating source cluster: ' + response['DBCluster']['DBClusterIdentifier'])
source_arn = (response['DBCluster']['DBClusterArn'])
create_target_cluster(target_cluster_name, source_arn, target_param_group_name)
response = rds.create_db_instance(
DBInstanceClass='db.r6g.2xlarge',
DBClusterIdentifier=source_cluster_name,
DBInstanceIdentifier=source_cluster_name + '-instance',
Engine='aurora-postgresql'
)
return(response)
def create_target_cluster(target_cluster_name, source_arn, target_param_group_name):
"""Creates a target Redshift cluster"""
response = redshift.create_cluster_parameter_group(
ParameterGroupName=target_param_group_name,
ParameterGroupFamily='redshift-1.0',
Description='For Aurora PostgreSQL zero-ETL integrations'
)
print('Created target parameter group: ' + response['ClusterParameterGroup']['ParameterGroupName'])
response = redshift.modify_cluster_parameter_group(
ParameterGroupName=target_param_group_name,
Parameters=[
{
'ParameterName': 'enable_case_sensitive_identifier',
'ParameterValue': 'true'
}
]
)
print('Modified target parameter group: ' + response['ParameterGroupName'])
response = redshift.create_cluster(
ClusterIdentifier=target_cluster_name,
NodeType='ra3.4xlarge',
NumberOfNodes=2,
Encrypted=True,
MasterUsername='username',
MasterUserPassword='Password01**',
ClusterParameterGroupName=target_param_group_name
)
print('Creating target cluster: ' + response['Cluster']['ClusterIdentifier'])
# Retrieve the target cluster ARN
response = redshift.describe_clusters(
ClusterIdentifier=target_cluster_name
)
target_arn = response['Clusters'][0]['ClusterNamespaceArn']
# Retrieve the current user's account ID
response = sts.get_caller_identity()
account_id = response['Account']
# Create a resource policy specifying cluster ARN and account ID
response = redshift.put_resource_policy(
ResourceArn=target_arn,
Policy='''
{
\"Version\":\"2012-10-17\",
\"Statement\":[
{\"Effect\":\"Allow\",
\"Principal\":{
\"Service\":\"redshift.amazonaws.com\"
},
\"Action\":[\"redshift:AuthorizeInboundIntegration\"],
\"Condition\":{
\"StringEquals\":{
\"aws:SourceArn\":\"%s\"}
}
},
{\"Effect\":\"Allow\",
\"Principal\":{
\"AWS\":\"arn:aws:iam::%s:root\"},
\"Action\":\"redshift:CreateInboundIntegration\"}
]
}
''' % (source_arn, account_id)
)
return(response)
def wait_for_cluster_availability(*args):
"""Waits for both clusters to be available"""
print('Waiting for clusters to be available...')
response = rds.describe_db_clusters(
DBClusterIdentifier=source_cluster_name
)
source_status = response['DBClusters'][0]['Status']
source_arn = response['DBClusters'][0]['DBClusterArn']
response = rds.describe_db_instances(
DBInstanceIdentifier=source_cluster_name + '-instance'
)
source_instance_status = response['DBInstances'][0]['DBInstanceStatus']
response = redshift.describe_clusters(
ClusterIdentifier=target_cluster_name
)
target_status = response['Clusters'][0]['ClusterStatus']
target_arn = response['Clusters'][0]['ClusterNamespaceArn']
# Every 60 seconds, check whether the clusters are available.
if source_status != 'available' or target_status != 'available' or source_instance_status != 'available':
time.sleep(60)
response = wait_for_cluster_availability(
source_cluster_name, target_cluster_name)
else:
print('Clusters available. Ready to create zero-ETL integration.')
create_integration(source_arn, target_arn)
return
def create_integration(source_arn, target_arn):
"""Creates a zero-ETL integration using the source and target clusters"""
response = rds.create_integration(
SourceArn=source_arn,
TargetArn=target_arn,
IntegrationName='my-integration'
)
print('Creating integration: ' + response['IntegrationName'])
def main():
"""main function"""
create_source_cluster(source_cluster_name, source_param_group_name)
wait_for_cluster_availability(source_cluster_name, target_cluster_name)
if __name__ == "__main__":
main()
Próximas etapas
Com um cluster de banco de dados do Aurora de origem e um data warehouse de destino do Amazon Redshift, agora você pode criar uma Integração ETL zero e começar a replicar dados. Para obter instruções, consulte Criar integrações ETL zero do Amazon Aurora com o Amazon Redshift.
