DEV Community: Jonas Barros

Change data capture com AWS RDS e PostgreSQL

Jonas Barros — Wed, 08 Jan 2025 23:41:50 +0000

Visão geral

Change Data Capture ou CDC é um método utilizado para extrair dados de uma base primária para uma outra base de destino utilizando algum conector para viabilizar essa comunicação

Esse conector será ativado quando ocorrer um evento no qual alterará o estado da sua base de dados primária.

Após a ocorrência deste evento, o conector será ativado recebendo os dados que foram alterados, com estes dados em mãos nós podemos enviá-los para outro ambiente, como por exemplo uma outra base de dados como um data lake.

Existem diversas ferramentas que poderíamos escolher para desenvolvermos o nosso fluxo de change data capture. O Dynamodb streams e AWS RDS com SQL Server são algumas ferramentas que possuem a funcionalidade de captura de eventos e migração de dados de forma nativa. Mas também podemos criar o nosso próprio change data capture utilizando os triggers do PostgreSQL para capturar as mudanças de estados e enviá-las ao conector.

Portanto ao decorrer deste artigo, irei mostrar como poderemos criar o nosso próprio fluxo de change data capture utilizando o AWS RDS Aurora PostgreSQL. Criaremos os triggers para observarmos as mudanças de estado e utilizaremos a extensão aws_lambda para enviar os dados ao conector AWS lambda.

Quando utilizar

Se a base de dados primária for um data lake, podemos criar um fluxo para notificar o conector a cada novo item que for inserido na tabela. O conector poderá estruturar os dados e enviá-los ao data warehouse ou data lake.
O dynamodb possui um recurso chamado dynamodb streams, com ele nós conseguimos notificar outros serviços da AWS quando houver uma alteração de estado em algum item da tabela.
Podemos instalar a extensão aws_lambda no AWS RDS Aurora PostgreSQL para noticiar um conector como por exemplo um AWS lambda quando o estado de alguma tabela for alterada.
Se tivermos implementando logs de auditoria, podemos aproveitar os recursos dos triggers do PostgreSQL para notificar os conectores quando houver um evento de inserção (insert), Atualização (update) ou Exclusão (delete) em uma ou múltiplas linhas da tabela.

Quando não utilizar

Ao utilizar para construção de simples tarefas, como envio de e-mail ou para outras notificações a cada momento que o estado é atualizado.
Quando a sua base de dados é muito pequena para se fazer essas transferências de dados, já que para construção deste fluxo acarretará em um aumento do custo mensal da conta AWS.
Se você possuir uma base em uma versão do RDS que não é compatível com a extensão aws_lambda. Portando não será necessário migrar toda a sua base dados ao AWS RDS Aurora, é possível utilizar outras estratégias para criar um fluxo de dados CDC.

Criando um change data capture utilizando um cluster AWS RDS Aurora PostgreSQL

Nos tópicos a seguir será apresentado como criar a infraestrutura inicial para um fluxo CDC no AWS Aurora PostgreSQL. Segue o link do repositório contendo todos os recursos necessários para iniciar a construção.

Repositório: Change data capture com AWS RDS Aurora PostgreSQL

Criando e configurando o cluster AWS RDS PostgreSQL

Utilizando o módulo aurora_postgresql_v2 nós podemos começar a criação de um cluster e uma instância do aurora PostgreSQL.
Para que não haja custos elevados, sua configuração foi a mais básica possível, evitando adicionar outros recursos como escalabilidade e segurança no cluster RDS.

Como nós criamos o cluster dentro de uma VPC, precisaríamos adicionar novos recursos para permitir a conexão entre máquina local e instância RDS. Um exemplo muito comum é criar uma instância EC2 que irá fazer a ponte e permitir a conexão entre máquina local e instância RDS. Porém fazer esse processo também envolve custos,
portanto para que não tenhamos custos elevados enquanto estamos estudando o CDC, nós criamos a instância de forma pública

vpc_id                 = "vpc-0000000" # adicione o ID VPC da sua conta AWS 
db_subnet_group_name   = "default-vpc-0000" # adicione a subnet da conta AWS
security_group_name    = "security" # adicione o security group name da conta AWS
publicly_accessible    = true # acesso público

Para realizar a conexão entre máquina local e cluster, podemos utilizar a porta 5432.

Para que a conexão entre lambda e cluster ocorra com sucesso, precisamos adicionar regras de entrada e saída. Dentro destas regras, adicionamos um range com IP padrão apenas para fins de demonstração, mas em um ambiente produtivo configure corretamente o range de IP no qual o cluster aceitará receber conexões.

A senha que precisamos para se conectar ao banco, é gerada automaticamente e adicionada no secret manager. Portanto,
para fazer a conexão com o cluster, acesse o secret manager no console AWS e copie a senha armazenada.

security_group_rules = {
  ex1_ingress = {
    cidr_blocks = ["0.0.0.0/0"]
    from_port = "5432"
    to_port = "5432"
    protocol = "tcp"
    type = "ingress"
  }

  ex2_ingress = {
    cidr_blocks = ["0.0.0.0/0"]
    from_port = "443"
    to_port = "443"
    protocol = "tcp"
    type = "ingress"
  }

  ex3_egress = {
    cidr_blocks = ["0.0.0.0/0"]
    from_port = "443"
    to_port = "443"
    protocol = "tcp"
    type = "egress"
  }

  ex4_egress = {
    cidr_blocks = ["0.0.0.0/0"]
    from_port = "5432"
    to_port = "5432"
    protocol = "tcp"
    type = "egress"
  }

Configurando o IAM

Como vimos no tópico anterior, foi preciso adicionar uma regra para aceitar conexões na porta 443 entre cluster e lambda.
Mas não é só isso, precisamos também configurar políticas e roles para que a conexão entre os dois serviços seja realizada com sucesso.

Criamos uma role do tipo Principal. Só com ela conseguiremos gerar credenciais temporárias para possibilitar a chamada da base de dados para a lambda

data "aws_iam_policy_document" "change-data-capture-document-role-cluster-rds" {
  statement {
    effect = "Allow"

    principals {
      type        = "Service"
      identifiers = ["rds.amazonaws.com"]
    }

    actions = ["sts:AssumeRole"]
  }
}

resource "aws_iam_role" "change-data-capture-role-cluster-rds" {
  name               = "change-data-capture-role"
  assume_role_policy = data.aws_iam_policy_document.change-data-capture-document-role-cluster-rds.json
}

Em seguida criamos uma política com uma ação de lambda:InvokeFunction, o objetivo é dizer ao cluster que ele está habilitado a fazer chamadas a nossa função lambda, que criaremos logo em seguida.

Com os recursos criados, agora só precisamos adicionar a role dentro do módulo postgreSQL

data "aws_iam_policy_document" "change-data-capture-ducument-policy-cluster-rds" {
  statement {
    sid = "ChangeDataCapturePolicy"
    effect = "Allow"
    actions = [
      "lambda:InvokeFunction"
    ]

    resources = [
      "arn:aws:lambda:us-east-1:AWS_ACCOUNT_ID:function:${var.lambda_name}"
    ]
  }
}

# anexamos a policy que criamos acima, dentro da nossa role
resource "aws_iam_role_policy_attachment" "change-data-capture-role-attachment-cluster-rds" {
  policy_arn = aws_iam_policy.change-data-capture-policy-cluster-rds.arn
  role = aws_iam_role.change-data-capture-role-cluster-rds.name
}

# dentro do modulo postgreSQL, adicionamos a role que criamos anteriormente
iam_roles = {
    lambda = {
      role_arn     = aws_iam_role.change-data-capture-role-cluster-rds.arn
      feature_name = "Lambda"
    }
  }

Criando a função lambda

Agora que terminamos a configuração do módulo postgresql e toda a estrutura de comunicação com outros serviços, precisamos criar e configurar a nossa lambda.

Precisei criar uma nova role do tipo Principal para geração de credenciais temporárias. Esse é o processo padrão para criação de qualquer lambda

data "aws_iam_policy_document" "change-data-capture-document-lambda-role" {
  statement {
    effect = "Allow"
    principals {
      type        = "Service"
      identifiers = ["lambda.amazonaws.com"]
    }
    actions = ["sts:AssumeRole"]
  }
}

resource "aws_iam_role" "change-data-capture-role-lambda" {
  name               = "change-data-capture-lambda-document"
  assume_role_policy = data.aws_iam_policy_document.change-data-capture-document-lambda-role.json
}

Em seguida, precisamos criar as políticas de acesso e a criação do recurso do cloud watch. Sem eles a lambda não conseguiremos observar o resultado da integração entre banco e lambda.

resource "aws_iam_policy" "change-data-capture-lambda-policy" {
  name        = "lambda_logging"
  path        = "/"
  description = "IAM policy for logging from a lambda"
  policy      = data.aws_iam_policy_document.change-data-capture-ducument-policy-cloud-watch.json
}

resource "aws_iam_role_policy_attachment" "change-data-capture-policy-attachment-lambda" {
  policy_arn = aws_iam_policy.change-data-capture-lambda-policy.arn
  role       = aws_iam_role.change-data-capture-role-lambda.name
}

resource "aws_cloudwatch_log_group" "change-data-capture-cloudwatch" {
  name              = "/aws/lambda/${var.lambda_name}"
  retention_in_days = 14
}

Em seguida precisaremos dizer ao terraform qual é o arquivo que será inserido na lambda. Para isso criei um recurso que me diz qual o caminho para o arquivo ZIP, no qual deverá conter todo código fonte que utilizaremos.
Segue o repositório contendo o código fonte desenvolvido em Javascript. Desenvolvemos algo bem simples, apenas enviamos ao CloudWatch o que a lambda está recebendo ao ser chamada pelo banco de dados.

data "archive_file" "change-data-capture-archive-file" {
  type        = "zip"
  source_dir  = "../src"
  output_path = "change-data-capture-function-payload.zip"
}

resource "aws_lambda_function" "change-data-capture-function" {
  filename      = "change-data-capture-function-payload.zip"
  function_name = var.lambda_name
  role          = aws_iam_role.change-data-capture-role-lambda.arn
  handler       = "main.handler"
  source_code_hash = data.archive_file.change-data-capture-archive-file.output_base64sha256
  runtime = "nodejs20.x"
  depends_on = [ 
    aws_iam_role_policy_attachment.change-data-capture-policy-attachment-lambda, 
    aws_cloudwatch_log_group.change-data-capture-cloudwatch
  ]
}

Conectando o AWS RDS com a função lambda

Antes de fazer os passos abaixo, certifique que os recursos foram criados corretamente quando rodou o comando terraform apply. Na AWS, acesse o secret manager e identifique a senha que foi criada quando o terraform foi executado. Copie e deixe-a separada já que iremos utilizá-la para se conectar com o banco.
A senha mudará a cada momento em que você fazer o destroy da infra e recriá-la novamente.

Conectando com o banco de dados

Na AWS acesse o RDS e identifique o cluster que foi criado. Ao selecionar o cluster, pegue a string de conexão, pois ela será necessária para fazermos a conexão entre a máquina local com o nosso banco de dados. A string de conexão deverá ter o seguinte formato:

{nome-do-cluster}.{numero-identificador}.{regiao}.rds.amazonaws.com

Para fazer a conexão com o banco, utilize o seguinte comando:

psql -h {nome-do-cluster}.{identificador-unico}.{regiao}.rds.amazonaws.com -U cdcStoreDevTo -p 5432 -d cdcstore

-h = host de conexão
-U = nome do usuário
-p = porta de conexão com o banco
-d = nome da database

Ao fazer o comando acima, vai aparecer no terminal um prompt para digitar a senha, portanto utilize a senha que foi criada e armazenada dentro do secret manager.
Se a conexão for realizada com sucesso, você estará neste momento dentro da instância cdcstore

Instalando a extensão aws_lambda

Após realizar a conexão com a instância, instale a extensão aws_lambda. Mas para que seja possível fazer a chamada, precisamos também instalar a extensão aws_commons.
Ao inserir o comando abaixo, será instalada ambas as extensões que precisaremos para enviamos dados ao aws lambda.

CREATE EXTENSION IF NOT EXISTS aws_lambda CASCADE;

Mas o que são extensões no postgreSQL?

A extensão é uma funcionalidade que você pode criar para fazer alguns trabalhos extras. Uma extensão bastante famosa no postgreSQL é o hstore, com ela nós podemos armazenar um conjunto de chave/valor em um único registro. Veja o exemplo abaixo, nós utilizamos a extensão como um tipo para o campo products, e adicionamos os dados chave/valor.

CREATE EXTENSION hstore;
CREATE TABLE store (products hstore);
INSERT INTO store VALUES ('a=>b, c=>d');

--consulta pelo item a
SELECT products['a'] from store;

-- consulta pelo item b
SELECT products['c'] from store;

A aws_lambda também é uma extensão, como dizemos anteriormente, as extensões vão fazer um trabalho extra, ou aprimorar alguma funcionalidade como vimos com hstore.
Veja que com o hstore é possível utilizar uma chave para consultar um único item ou um conjunto deles, dependendo dos valores que foram adicionados.

Também é possível criar as nossas próprias extensões no postgreSQL. Para isso podemos criar um arquivo com extensão nome_do_arquivo.sql, e adicionar todo o conteúdo da extensão dentro dele, e salvaremos esse arquivo dentro do diretório de instalação do postgreSQL na pasta /extension. Para encontrar o diretório de instalação consulte a documentação oficial do banco. Mas para sistemas operacionais baseados em linux, possivelmente irá encontrá-lo no seguinte caminho: /usr/share/postgresql/{NUMERO_DA_VERSAO_DO_POSTGRESQL}/extension

Após escrevemos a nossa extensão e inseri-la na pasta /extension, podemos instalar a extensão com o comando a seguir. Após instalada, poderá ser utilizada normalmente conforme você desenvolveu.

CREATE EXTENSION nome_extensão;

Porém como estamos trabalhando com RDS, não será possível acessarmos o diretório de instalação do postgreSQL para criarmos as nossas extensões, logo teremos que utilizar uma nova abordagem chamada TLE - Trusted Language Extensions.

Trusted Language Extension é um kit de desenvolvimento de código aberto utilizado para criar as nossas extensões de maneira segura. Como foi visto anteriormente, para instalar as extensões da maneira padrão, é necessário criar um script dentro da pasta /extension localizada no diretório de instalação do postgreSQL. Porém, por motivos de segurança o RDS não possibilita o acesso ao sistema de arquivos da máquina no qual está rodando a instância. E se desejarmos criar a nossa extensão para rodar dentro do RDS será necessário utilizar o kit de desenvolvimento TLE.

Quando instalamos o TLE, temos disponíveis o que chamamos de linguagens seguras para desenvolver os nossos scripts, como por exemplo o PL/V8 no qual é uma biblioteca que fornece uma linguagem procedural para PostgreSQL que utiliza o V8 JavaScript Engine. Ao utilizá-la nós não conseguiremos acessar ou manipular quaisquer conteúdos dentro do sistema de arquivos da instância que roda o banco de dados.

Na documentação do PostgreSQL temos um exemplo no qual seremos barrados ao tentarmos criar um arquivo dentro do diretório /tmp. Este exemplo foi feito utilizando o PL/PERL no qual também é uma versão segura para criarmos as nossas extensões.

CREATE FUNCTION badfunc() RETURNS integer AS $$
    my $tmpfile = "/tmp/badfile";
    open my $fh, '>', $tmpfile
        or elog(ERROR, qq{could not open the file "$tmpfile": $!});
    print $fh "Testing writing to a file\n";
    close $fh or elog(ERROR, qq{could not close the file "$tmpfile": $!});
    return 1;
$$ LANGUAGE plperl;

O comando a seguir pressupõe que você já tenha criado um grupo de parâmetros para a sua instância, então você só precisará substituir a variável ADD_PARAMETER_GROUP_NAME pelo nome do grupo de parâmetros. Ao final deste tópico terá um link para a documentação auxiliando na criação de um grupo de parâmetros para a instância que você está utilizando.

aws rds modify-db-parameter-group \
--db-parameter-group-name ADD_PARAMETER_GROUP_NAME \
--parameters"ParameterName=shared_preload_libraries,ParameterValue=pg_tle,ApplyMethod=pending-reboot" \
--region aws-region

No comando acima nós estamos dizendo ao RDS para inicializar a extensão pg_tle na instância que está vinculada ao grupo de parâmetros.

Se não houver nenhum erro ao executar o comando acima você deverá reiniciar a instância e logo em seguida baixar a extensão e executar o seguinte comando:

CREATE EXTENSION NOME_EXTENSAO;

Segue alguns links úteis no qual explicam com mais detalhes sobre o TLE e nos ensinam a desenvolver nossas primeiras extensões.

As extensões hstore e aws_lambda já vem pré-carregada dentro da instância aurora-postgresql portanto não será necessário realizar os descritos acima.

Testando funcionalidade

Vamos fazer um primeiro teste simples. Vamos utilizar a função create_lambda_function_arn e passar o ARN da nossa lambda, em seguida passaremos uma string contendo apenas uma simples mensagem e converteremos esse valor para o tipo json

SELECT * from aws_lambda.invoke(aws_commons.create_lambda_function_arn('ARN FUNCAO LAMBDA', 'us-east-1'), '{"body": "Hello Word"}'::json);

Agora vá a AWS Console e observe os logs da função lambda no cloud watch. Veja que o resultado apresentado lá dentro condiz com
o valor que passamos dentro da função aws_lambda.invoke.

Testando funcionalidade de uma maneira mais complexa

Vamos agora fazer algo mais complexo. Utilizaremos um trigger para ser acionado sempre quando houver uma inserção no banco de dados. Em seguida iremos notificar uma lambda que apenas escreverá no cloudWatch o valor que inserimos na função.

Criando uma tabela

Primeiro vamos criar uma tabela para que no momento que inserirmos os dados o nosso trigger seja notificado

CREATE TABLE store (
  store_id SERIAL PRIMARY KEY,
  name_store VARCHAR(100) NOT NULL,
  address VARCHAR(100) NOT NULL,    
  segments VARCHAR(100) NOT NULL
);

Criando a estrutura do trigger

Para se criar um trigger é muito simples então podemos seguir o que está descrito na documentação oficial

CREATE TRIGGER integration_lambda_trigger
  AFTER INSERT ON store
  FOR EACH ROW 
  EXECUTE FUNCTION integration_lambda()

-- integration_lambda_trigger: nome do trigger
-- integration_lambda(): nome da function

Em seguida precisamos criar uma função que será chamada quando o trigger for acionado.

CREATE FUNCTION integration_lambda() RETURNS TRIGGER AS $$
  DECLARE
    json_text TEXT;

  BEGIN 
    json_text := json_build_object('store_id', NEW.store_id, 'name_store', NEW.name_store, 'address', NEW.address, 'segments', NEW.segments);

    PERFORM aws_lambda.invoke(aws_commons.create_lambda_function_arn('arn:aws:lambda:us-east-1:700552527916:function:change-data-capture','us-east-1'), json_text::json);

    return NEW;

  END;
$$ LANGUAGE plpgsql;

Iniciamos a função com uma declaração de variável chamada json_text, nós utilizaremos para armazenar o valor de retorno da função chamada json_build_object(). Ela será a responsável por criar um JSON de acordo com os dados que inserimos na tabela store. Para isso, teremos que estruturar a função da seguinte maneira. Devemos passar o nome da chave do json e logo em seguida o seu respectivo valor.
Para pegarmos este valor podemos utilizar o operador NEW composto pelo nome da coluna da tabela. NEW.store_id

O próximo passo será fazer a chamada para a função Lambda com aws_lambda.invoke. Como primeiro parâmetro
da função temos que passar o endereço da nossa lambda, ou seja o ARN e a região no qual a lambda foi criada.
Já no segundo parâmetro, passamos a variável json_text no qual estará armazenado o JSON que enviaremos ao lambda.
Observe também a expressão ::json. Somos obrigados a passar esse valor porque temos que fazer um cast para deixar explícito o tipo de dado que estamos enviando.

Por fim, retornamos novamente o operador NEW no final da função.

Se observar no repositório, teremos um script SQL no qual temos todos esses passos sendo executados. Portanto vamos utilizá-lo enviando o comando no momento de fazer a conexão
com o banco de dados.

Mas também podemos utilizar o PgAdmim para executar todas as funcionalides incluidas no script.

psql -h {nome-do-cluster}.{identificador-unico}.{regiao}.rds.amazonaws.com -U cdcStoreDevTo -p 5432 -d cdcstore -f "localizacão do script .SQL"

Lembre-se que a senha para acesso ao banco se encontra no secret manager.

Se você acabou de criar sua a sua instância e executou o script .sql, certamente você não terá nenhum recurso criado, logo aparecerá no seu terminal alguns logs de erro
como esses:

.../sql/script.sql:1: ERROR: relation "store" does not exist

.../sql/script.sql:2: ERROR: could not find a function named "integration_lambda"

.../sql/script.sql:3: ERROR: table "store" does not exist

.../sql/script.sql:5: NOTICE: extension "aws_lambda" already exists, skipping

Mas não se preocupe, esses erros só ocorrerão quando você executar pela primeira vez, já que a todo o momento que executamos o script, adicionei algumas funcionalidades para remover todos os recursos criados anteriormente e recriá-los novamente.

A seguir podemos executar alguns inserts para visualizar todos os recursos em funcionamento.

INSERT INTO store (store_id, name_store, address, segments) VALUES (1, 'my store', '76 street', 'tecnology');

INSERT INTO store (store_id, name_store, address, segments) VALUES (2, 'my store', '76 street', 'tecnology');

Se os comando retornarem sucesso após a sua execução, podemos olhar no CloudWatch e verificarmos o que a função lambda recebeu

2024-12-18T20:01:12.517Z    INFO    {
  store_id: 1,
  name_store: 'my store',
  address: '76 street',
  segments: 'tecnology'
}

2024-12-18T20:01:19.034Z    INFO    {
  store_id: 2,
  name_store: 'my store',
  address: '76 street',
  segments: 'tecnology'
}

Portanto se você visualizou esse resultado, o fluxo funcionou como deveria. Como expliquei no inicio deste artigo, existem diversas maneiras no qual você utilizar para sair com o mesmo resultado. E também existem diversos casos de uso no qual fará sentido utilizar esses recursos, então analise com cuidado e opte ou não pela criação de um change data capture.

Streaming de vídeos com AWS MediaConvert

Jonas Barros — Sun, 04 Aug 2024 14:22:35 +0000

Prólogo

Neste artigo, apresentarei como foi o processo de discovery e desenvolvimento do tech challenge que teve o seguinte tema: desenvolvimento de uma aplicação web de streaming de vídeos. Veremos quais ferramentas foram utilizadas, entenderemos como funciona um streaming de vídeo, aprenderemos sobre codecs, bitrate, segmentos, protocolos de vídeo, e, por fim, integraremos todas essas tecnologias utilizando o AWS MediaConvert.

Esse tech challenge é um dos quatro trabalhos que tivemos que desenvolver durante o período da pós-graduação realizada na FIAP.

Ferramentas da AWS utilizadas para desenvolvimento do projeto

AWS S3: Para armazenar as mídias originais, e os pequenos pacotes de vídeo gerados após o processamento do MediaConvert.
AWS MediaConvert: Este é um serviço específico para processamento de vídeos. Com ele, é possível definir os protocolos de reprodução de vídeos, segmentos, bitrate, resolução, codecs e diversos outros parâmetros utilizados no desenvolvimento deste projeto.
AWS SQS: O serviço de mensageria da AWS foi utilizado para fazer a comunicação entre os dois microsserviços. O primeiro microsserviço é chamado de stream de vídeos, enquanto o segundo é chamado de pipeline de processamento de vídeos. Ao longo deste artigo, focaremos no desenvolvimento do microsserviço chamado pipeline de processamento de vídeos.

Arquitetura do projeto

Abaixo estão os dois microsserviços que precisaram ser criados para o desenvolvimento do techchallange. O serviço que iremos aprofundar para melhor entendimento é a pipeline de processamento de vídeos. Dentro deste microsserviço nós temos a integração com o MediaConvert no qual será o foco principal deste artigo.

O primeiro microsserviço se chama stream de vídeos, será o responsável por enviar a mídia original ao AWS S3 e fazer as outras regras de negócio que eram requisitos obrigatórios do tech challange.

Já o segundo microsserviço se chama pipeline de processamento de vídeos, em meu cenário ele é o responsável por fazer o processamento dos vídeos, por exemplo, separar a mídia bruta em pequenos pacotes para agilizar o tráfego do conteúdo na rede do usuário, definição dos codecs, resolução e bitrate.

O MediaConvert no qual utilizamos, é uma ferramenta extremamente poderosa, com ela nós conseguimos fazer diversas configurações, inclusive fazer pequenas melhorias no vídeo. Se não conhecer o MediaConvert, é importante entender um pouco sobre ele antes de continuar a leitura. Foi com ele que fizemos toda a configuração para processamento de vídeos.

Observe também que entre os dois microsserviços existe o AWS SQS, como disse anteriormente ele é responsável por fazer a comunicação entre os serviços. No payload do SQS enviaremos o nome da mídia para que a pipeline de processamento de vídeo encontre-a no bucket e inicie os trabalhos de processamento.

Existem outras ferramentas que poderíamos utilizar para construção de ambos os microsserviços. Uma das possibilidades seria a inclusão do AWS CloudFront para acelerar a entrega do conteúdo já processado em pequenos pacotes e armazenado no AWS S3. Mas todas essas mudanças e melhorias ficariam para uma segunda versão. A versão atual além de funcionar perfeitamente bem, fez com que o problema fosse resolvido de forma simples e o projeto tenha sido entregue dentro do prazo estipulado.

Quais tecnologias utilizamos para processamento de vídeo?

Antes de iniciar qualquer desenvolvimento da pipeline de processamento de vídeo, precisaria estudar algumas ferramentas e configurações de edição de vídeo no qual eu não estava completamente familiarizado. Esta etapa foi importante para entender cada uma das ferramentas antes de partir para o desenvolvimento.

Portanto Tive que responder algumas perguntas como por exemplo; o que seriam os protocolos HLS, IIS ou Dash ISO? O que significa bitrate? Baseado em meu cenário, qual o melhor codec de vídeo que eu poderia escolher? Me aprofundei para entender cada dos questionamentos e como isso poderia trabalhar de forma conjunta dentro de uma aplicação.

O que significam os protocolos HLS, IIS e Dash ISO?

HLS (HTTP Live Stream): Lançado em 2009 pela Apple, em princípio para funcionar apenas em seus Iphones porém logo se espalhou pelo mercado e atualmente é o protocolo mais utilizado para reprodução de vídeos. As principais características do HLS é a fragmentação da mídia original em pequenos pacotes com diversas resoluções para rápida distribuição de conteúdo pela rede do usuário. Para isso ele utiliza um formato chamado m3u8, no qual é um arquivo utilizado pelos players de reprodução de vídeo para que eles consigam saber a lista de reprodução para cada resolução.

O protocolo HLS funciona da seguinte forma: Primeiro teremos que ter a mídia original, essa mídia deve estar com uma excelente qualidade de vídeo, a resolução ideal é estar acima do 1080p.
A mídia será dividida em centenas ou milhares de pacotes (iremos definir isso dentro da aplicação), e cada uma deles terá uma resolução específica podendo variar entre 480p, 720p ou 1080p ou nas outras resoluções que você definiu. Após dividir a mídia original em pacotes menores, será criado um arquivo chamado de m3u8 que dentro dele terá uma lista de reprodução para cada resolução. Ou seja, terá uma "link" para a lista de reprodução da resolução 1080p, 720p, 480p assim sucessivamente. Portanto os reprodutores utilizaram esse arquivo m3u8 para renderizar o vídeo em seu aparelho.

No momento é isso é tudo que você precisará saber. Na segunda parte irei entrar em detalhes como funciona o HLS e descrever cenários práticos para melhor entendimento. Logo já adiantando que para desenvolvimento desse projeto, utilizei como protocolo de vídeo o HLS. Mas leia também os outros protocolos que poderiamos ter utilizado
IIS (Live Smooth Stream): O protocolo de transporte smooth streaming desenvolvido pela Microsoft, também é uma tecnologia de streaming adaptável, e utiliza o mesmo conceito do Apple HLS, ou seja utiliza-se o HTTP para transmitir suas mídias, ao player de destino. Portanto ele é bem semelhante ao HLS, porém com um diferencial, na época em que foi criado, tinha suporte a plataforma de reprodução de vídeo chamada Silverlight, no qual foi descontinuada anos depois.
Dash ISO: É um padrão desenvolvido pela International Organization for Standardization (ISO), esse padrão não é muito diferente dos outros dois, mantendo as mesmas características de comunicação com protocolo HTTP, reprodução adaptável entre outras características.

Baseado em meu cenário, quais os melhores codecs de vídeo que poderiamos utilizar?

Para responder essa pergunta, primeiro precisamos entender como funcionam os codecs de vídeo.

Os codecs utilizam algoritmos para comprimir a mídia em um formato compacto. O codec que utilizei para desenvolver o projeto é atualmente um dos mais populares, chamado de H-264. Ele é amplamente utilizado no mercado, porém se tivéssemos construindo o nosso próprio reprodutor de vídeo e optassemos por utilizar este codec, teríamos que adquirir uma licença de uso. Porém existem outros codecs de código aberto disponíveis no mercado como por exemplo o AV1 desenvolvido pela Alliance For Open Media

Para desenvolvimento do tech challange, estamos utilizando o AWS MediaConvert portanto não precisamos adquirir a licença de uso, pois a AWS tratou de todos esses assuntos e só precisamos utilizá-lo, sem se preocupar com esses assuntos.

Para comprimir uma mídia são utilizados algoritmos que usam as técnicas de entropia ou entropy method. O CAVLC e CABAC são alguns desses algoritmos que utilizam as técnicas de entropia para comprimir um arquivo de mídia. Para não estender muito este artigo explicando como funcionam esses algoritmos, irei deixar o link de um outro artigo que ao ler me ajudou muito no entendimento do CAVLC.

codificação de comprimento variável adaptável ao contexto

Outro ponto importante que precisamos entender é que não é trabalho dos codecs fazerem a fragmentação da mídia original, eles apenas comprimem a mídia já fragmentada para melhor distribuição do conteúdo na rede do usuário. Quem divide a mídia em multiplos pacotes menores são os protocolos de vídeo

O que é BitRate?

A configuração do bitrate é essencial para termos uma boa experiência na qualidade da imagem, pois quanto mais baixo, pior será a apresentação do vídeo, porém quanto mais alto melhor a apresentação ou seja melhor será a qualidade da imagem. Para definirmos o melhor bitrate não precisamos fazer cálculos complexos, o youtube tem uma documentação com os bitrate definidos para cada resolução. Segue a documentação disponibilizada pelo mesmo.

Para finalizarmos toda a teoria ainda precisei entender como funciona o AWS MediaConvert no qual é a ferramenta que utilizei para fazer o processamento dos vídeos. Para isso tive que acessar o AWS Skill Builder e fazer alguns cursos que explicavam como utilizar o serviço diretamente na console e também cada configuração que precisaríamos fazer, começando por enviar a mídia original, escolha do codec H-264, definindo os parâmetros de configuração de áudio e vídeo, e iniciando o processamento da mídia original e separando em pequenos pacotes e armazenando em um bucket no S3.

Segue os cursos que foram feitos no AWS Skill Builder

AWS Skill Builder - Setting Up and Configuring AWS Elemental MediaConvert
AWS Skill Builder - AWS Elemental MediaConvert Primer
AWS Skill Builder - Creating Adaptive Bitrate (ABR) Outputs

Após responder todas as questões e aprofundar sobre cada tema, chegou a hora de começarmos o desenvolvimento.

Desenvolvimento do projeto

Para integração com o MediaConvert temos que utilizar um SDK da AWS, para isso estou utilizando a versão 2.x do SDK da AWS.

Antes de iniciar o desenvolvimento, vamos analisar o código apresentado na documentação no item "Create Job". Veja que já conseguimos entender como fazer as configurações do áudio, a resolução que gostaríamos que as nossas mídias tenham, configuração da taxa de bitrate e codec utilizado. Ou seja, nós temos toda a base necessária para iniciarmos os trabalhos.

Observe que é um código apresentado na documentação é bastante extenso, porém no tech challenge tive que desenvolver o meu projeto baseado no que já estava disponibilizado na documentação da AWS. Já fiz o trabalho de separar cada recurso em pequenos "blocos" e adicionar as configurações necessárias para que tudo funcione conforme o esperado. Portanto disponibilizo o repositório para que possamos acompanhar e entender onde devemos encaixar cada configuração de vídeo explicada no capítulo "Quais tecnologias utilizamos para processamento de vídeo?"

O repositório, faz parte da pipeline de processamento de vídeo, nele você encontrará além da integração com o MediaConvert, também verá a integração com o S3 para que possamos pegar a média original e enviá-la para o MediaConvert, encontrará também a integração com o SQS, pois é a partir das mensagens que recebemos da fila que iniciamos a nossa pipeline de processamento de vídeo.

Para cada mídia enviada ao MediaConvert, precisará ter configurada a trilha de áudio, porém como o nosso foco não é configuração de áudio, vamos deixar da forma padrão conforme visualizamos na documentação.

https://github.com/JonasBarros1998/processamento-de-videos/blob/main/src/main/java/com/br/stream/processamentoDeVideo/infra/processarMidias/audio/TrilhasDeAudio.java

Vamos configurar os codecs de vídeo, como já foi explicado no capítulo anterior, utilizei o codec H-264, mas dentro do MediaConvert também temos disponíveis outros codecs como Dash ISO, CMAF e Microsoft Smooth Streaming.

A configuração do codec é bastante extensa, por enquanto não precisamos nos preocupar com todos os parâmetros de configuração, mas um dica valiosa que podemos seguir é entrarmos na console da AWS ir até a ferramenta MediaConvert, adicionar um grupo de saída e ao selecionar Apple HLS, veremos uma série de parâmetros a serem configurados, se observarmos os parâmetros na console AWS e observarmos os parâmetros descritos no código veremos muita semelhança, assim facilitando o nosso desenvolvimento, e o entendimento de cada item.

Para criar o container Apple HLS: AppleHLSContainer
Para configurar o segmento de cada pequeno "bloco" da mídia principal: Nesse caso utilizei um segmento igual a 10. Isso quer dizer que cada pacote deverá ter um tamanho de 10 segundos no total. Então o MediaConvert pega a mídia original divide ela em pequenos pacotes de 10 segundos para cada uma da resolução escolhida. Essa configuração está dentro da classe AppleHLSContainer. Configuração do segmento
Taxa de bits ou bitrate: Para configurar a taxa de bits ideal, conforme disse no capítulo anterior, utilizei as mesmas configurações do youtube para as resoluções 480p, 720p e 1080p. Definição do BitRate e configuração para resoluções 480p, 720p e 1080p
Configuração do codec H-264: Codec H-264
Outras configurações de áudio inseridas dentro do codec: Container de áudio

Portanto, este foi o trabalho desenvolvido para a fase 4 do tech challange da Fiap. Veja que para chegarmos na fase de desenvolvimento tive que entender diversas tecnologias que até momentos atrás não faziam parte do meu escopo de conhecimento. Portanto agora conseguimos entender melhor como o youtube, netflix e outros players de vídeo fazem para reproduzir aquela série que tanto gostamos em nossos aparelhos celulares ou nos computadores. Observe também que só utilizamos ferramentas AWS para salvar as nossas mídias, processá-las e entregá-las ao usuário, a utilização da AWS foi um dos pontos principais para que eu consiga ter agilidade no desenvolvimento, pois é uma nuvem no qual já tenho experiência portanto o desenvolvimento do projeto foi um pouco mais rápido.

Estratégias de cache | Cache Strategies

Jonas Barros — Wed, 26 Jun 2024 00:20:24 +0000

Estratégias de armazenamento em cache; write-through, read-through, lazy loading e TTL

Write-Through: Essa estratégia é utilizada quando precisamos de consistência dos dados entre o meu banco de dados primário (SQL) e o cache. Para sua implementação correta, precisamos adicionar os dados no cache em todos os métodos que houver uma chamada para escrita de dados relacionada ao banco de dados principal.

Por exemplo, se em uma aplicação eu tiver uma funcionalidade para cadastrar um cliente, após esse cadastro ser concluído, devemos adicionar esses dados também no cache.

Portanto, uma das vantagens de utilizar essa estratégia é que eu sempre terei os dados atualizados, podendo assim sempre fazer a consulta diretamente no meu cache, ao invés de ir ao banco de dados, assim melhorando o tempo de resposta de uma consulta.

Sempre teremos os dados atualizados no cache e não iremos ter eventuais inconsistências entre cache e base de dados.
Quando precisarmos realizar uma consulta, podemos ir direto no cache ao invés de ir no banco de dados, resultando assim em um tempo reduzido para consultas e menor tráfego de acesso ao meu banco de dados.

Umas das desvantagem da utilização do padrão write-though, é o crescimento rápido do armazenamento dos dados em cache, já que se você adicionar no cache dados de toda e qualquer escrita do banco de dados primário, os dados menos acessados poderão ocupar muito espaço, assim inflando o armazenamento.
Adicionar uma TTL e uma boa estratégia de lazy loading, poderá contornar esse problema acima.

function insertDataOnDatabase() {
  // add logic of the insertion 
}

function insertDataOnCacheDatabase() {
  // add logic of the insertion
}

function updateClientOnDatabase() {
  // add logic for update
}

function updateClientOnCacheDatabase() {
  // add logic for update
}


function insertNewClient() {
  //should insert data into primary database
  insertDataOnDatabase();

  //insert data into database cache
  insertDataOnCacheDatabase();
}

function updateClient() {

  //should update client into primary database
  updateClientOnDatabase();

  //update client into database cache
  updateClientOnCacheDatabase();

}

Lazy Loading: Essa estratégia é utilizada quando queremos adicionar aqueles dados que são mais acessados nas aplicações. Para isso temos que implementar essa estratégia da seguinte forma:

O usuário solicita os dados de uma aplicação.
A aplicação vai primeiro consultar os dados no cache.
Se esses dados não existirem, a aplicação vai consultar esses dados no banco.
Em seguida os dados serão adicionados no armazenamento do cache.

Se em uma segunda vez o usuário solicitar esses mesmos dados, não precisamos buscar no banco de dados primário, agora eu busco diretamente no meu cache. Trazendo assim agilidade na consulta dos dados.

Vantagens:

A aplicação só irá salvar os dados que realmente utilizamos.
Agilidade na hora da consulta dos dados.
Menos acessos ao meu banco de dados primário.

Desvantagens:

Se o dado não existir no cache, tenho que passar por 4 passos (conforme listados acima) até retornar os dados ao usuário, gerando assim uma demora maior para retorná-los. Para evitar que todos esses passos sejam executados em sequência, atualize os dados do cache de maneira assíncrona.
Dados inconsistentes no cache. Como o cache só irá ser atualizado quando houver alguma consulta, e se por algum momento esse dado ser atualizado no banco e não ser atualizado no cache, isso gerará uma inconsistência entre banco de dados primário e cache. Se você precisa garantir a consistência de dados entre cache e banco de dados primário, implemente também a estratégia de write-through. Porém se a inconsistência não for um problema, siga em frente com essa estratégia sem maiores problemas.

function findDataOnDatabase() {
  // add logic of the insertion 
}

function insertDataOnCacheDatabase() {
  // add logic of the insertion
}

function updateClientOnCacheDatabase() {
  // add logic for update
}

function findDataOnCacheDatabase() {
  // add logic for search 
}


function findClient() {

  const client = findDataOnCacheDatabase()

  if(client === null) {

    const dataOfClient = findDataOnDatabase();

    insertDataOnCacheDatabase();

    return dataOfClient;
  }

  return client;

}

Referências

O que é ACID?

Jonas Barros — Mon, 04 Jul 2022 23:25:56 +0000

O termo foi criado no ano de 1983, por Andreas Reuter e Theo Harder, no qual visa definir os 4 termos que compõem uma transação em um banco de dados. Atomicidade, Consistência, Isolamento e Durabilidade.

O que é uma transação?

Uma transação é uma única unidade de trabalho contendo alguma operação do banco de dados como por exemplo o INSERT. Quando iniciamos uma transação, e se por algum motivo uma ou mais operações falharem, nós poderemos fazer o rollback ou seja voltar o banco ao seu estado anterior. Veja o exemplo abaixo:

1: CREATE SCHEMA acid;

2: CREATE TABLE acid.herois (
3:  nome VARCHAR(255) NOT NULL,
4:      sobrenome VARCHAR(255) NOT NULL,
5:      UNIQUE(nome, sobrenome)
6: );
7:
8: START TRANSACTION;
9: INSERT INTO acid.herois values('Steve', 'Rogers'), ('Tony', 'Stark');
10: COMMIT;

Repare na linha 8, é nesse ponto que iniciamos uma transação. Se a operação feita na linha 9 falhar o commit não será executado e teremos que fazer o rollback.

Implementando na prática os 4 termos do ACID.

Antes de partir para a prática, vamos entender os 4 termos do ACID.

Atomicidade: Todas as operações incluídas dentro de uma transação devem ser realizadas com sucesso, se por acaso alguma operação falhar o rollback deverá ser feito e o banco voltará ao seu estado anterior. Porém se todas as operações forem efetuadas com sucesso a transação deve ser concluída com o comando COMMIT.

Consistência: Todas as regras e restrições que definimos no banco de dados devem ser obedecidas, como por exemplo, tipo dos campos (Boolean, Char, Varchar, Integer, Date, etc), ou seja eu não posso adicionar um texto em um campo onde seu tipo é Integer, relacionamentos por chaves estrangeiras, ou seja todas essas regras devem ser obedecidas.

Isolamento: Nenhuma transação poderá interferir no funcionamento da outra transação se as duas estiverem rodando em paralelo. E nenhuma transação poderá visualizar os resultados parciais das outras transações. Ou seja, cada transação tem que ser completamente independente, entenderá melhor esse funcionamento mais adiante, porém já adiantando que essa regra irá mudar de acordo com o tipo do SGBD que você utilizará.

Durabilidade: O resultado de uma transação são permanentes e só poderão ser alterados por uma transação subsequente.

Atomicidade

Por padrão a instrução INSERT já é atômica, ou seja, eu não preciso adicionar explicitamente o comando START TRANSACTION como fizemos acima, para dizer que vou iniciar uma transação. Tente executar esses dois inserts abaixo de forma separada no seu editor de SQL e veja o que acontece.

CREATE TABLE banco (
    nome VARCHAR (50) NOT NULL,
    saldo MONEY NOT NULL
);

INSERT INTO banco (NOME, SALDO) VALUES ('Tim', 100); 

INSERT INTO banco (NOME, SALDO) VALUES ('Tim');

Veja que o primeiro INSERT tudo funcionou, porém já no segundo INSERT ocorreu um erro porque estamos tentando adicionar um valor NULL em uma coluna NOT NULL, logo o postgresql não vai adicionar apenas um valor no banco se caso o segundo valor a ser adicionado esteja incorreto.

Veja agora esse segundo exemplo.

1: DO
2: $$
3: DECLARE 
4:  I INT;
5: BEGIN
6:    FOR I IN 1..3 LOOP
7:      INSERT INTO banco (NOME, SALDO) VALUES ('Tim', 100);
8:  END LOOP;
9: END; 
10: $$;
11: 
12: DO
13: $$
14: DECLARE 
15: I INT;
16: BEGIN
17:    FOR I IN 1..3 LOOP
18:     INSERT INTO banco (NOME, SALDO) VALUES ('Martin', 300);
19: END LOOP;
20: END; 
21: $$;

No código acima, fizemos um laço com 3 repetições, e dentro dele temos uma simples instrução INSERT.

Após executar o código acima, veja que os valores foram adicionados corretamente e todas as instruções passaram sem maiores problemas.
Agora apague tudo que foi adicionado na tabela, porque vamos forçar um erro no segundo INSERT na linha 18 para ver o que acontece.

1: DO
2: $$
3: DECLARE 
4:  I INT;
5: BEGIN
6:    FOR I IN 1..3 LOOP
7:      INSERT INTO banco (NOME, SALDO) VALUES ('Tim', 100);
8:  END LOOP;
9: END; 
10: $$;
11: 
12: DO
13: $$
14: DECLARE 
15: I INT;
16: BEGIN
17:    FOR I IN 1..3 LOOP
18:     INSERT INTO banco (NOME, SALDO) VALUES ('Martin');
19: END LOOP;
20: END; 
21: $$;

Veja que apaguei o saldo da conta do Martin, porém a coluna saldo não aceita valores do tipo NULL.

Ao executar esse bloco de código repare houve um erro na linha 18, onde estamos tentando inserir um valor do tipo NULL em uma coluna que não permite tais valores, porém se rodarmos o comando SELECT * FROM banco; veja que nenhum valor foi adicionado, nem mesmo os valores contidos no INSERT da linha 7. Então podemos afirmar que esse conjunto de instruções são atômicas, porque toda a instrução tem que ser bem-sucedida para que os dados sejam adicionados ao banco, se por acaso alguma instrução falhar nada será modificado em nossas tabelas.

Isso acontece pelo seguinte motivo, repare na linha 16 temos a instrução chamada BEGIN essa instrução serve para iniciarmos uma transação, e o COMMIT para confirmar. Mas repare que eu não adicionei nenhuma instrução COMMIT ao código, porque especificamente no postgresql quando se inicia uma transação e não existe uma instrução COMMIT ele executa a transação em modo de auto commit veja com mais detalhes na documentação.

Consistência

O exemplo acima explica bem como funciona a consistência do banco de dados. Repare que não conseguimos adicionar um valor do tipo NULL em uma coluna do tipo MONEY. Também não é possível adicionar uma palavra qualquer ao invés de um valor monetário.
INSERT INTO banco (NOME, SALDO) VALUES ('Martin', 'money');

Durabilidade

Toda a transação é permanente e só pode ser desfeita por uma outra transação.
Após o COMMIT de uma transação os dados devem ser salvos ou excluídos de forma definitiva no banco de dados, não podendo haver erros por falha de hardware por exemplo.

Isolamento

No postgresql existem 4 tipos de isolamento; Read Uncommitted, Read Committed, Repeatable Read, Serializable, mas por padrão ele utiliza o Read Committed. Com o comando SET TRANSACTION podemos alterar esse comportamento.
Para fins didáticos vamos entender apenas o tipo Read Committed.

read commited
Esse nível de isolamento funciona da seguinte maneira, por exemplo o comando SELECT ele vai ler apenas os dados que foram confirmados. Isso acontece porque o SELECT lê um snapshot do banco de dados antes do início da leitura. O comando SELECT também pode ler as alterações que foram feitas dentro da mesma transação, mesmo que não tenham sido confirmadas. Veja o exemplo abaixo. (Ainda não execute a instrução COMMIT).

START TRANSACTION;
    INSERT INTO banco (nome, saldo) VALUES ('Mark', '200');
    SELECT * FROM banco;
/*COMMIT*/

Veja que o SELECT leu os dados que ainda não foram confirmados.

Em seguida, abra mais uma instância do seu editor de SQL e faça uma nova conexão com o banco, depois digite a instrução SQL abaixo, para verificarmos se ele conseguirá ler as informações que adicionamos acima, porém ainda não confirmadas.

SELECT * FROM banco;

Veja que nada nos retornou. Agora descomente e execute a instrução COMMIT que fizemos acima, e posteriormente rode a instrução SELECT que fizemos fora da transação. Tudo funcionou como esperado, após confirmamos a transação os dados foram adicionados no banco e o select que estava fora da transação conseguiu ler os dados que foram inseridos na tabela.

Mas se tivermos 2 transações em andamento de forma paralela, a primeira transação excluindo linha com o nome Mark já a segunda atualizando essa mesma linha. O Read Committed irá tratá-las da seguinte maneira.
Se temos duas transações rodando em paralelo, se a primeira transação for uma exclusão, a segunda transação irá aguardar a confirmação dos resultados da primeira transação. Se a primeira transação for confirmada, a segunda transação irá ignorar a linha excluída pela primeira transação. Porém se as atualizações forem revogadas a segunda transação irá prosseguir normalmente com as suas atualizações.

Referências:

Transactions
Transacions Isolation