As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
# Solução de problemas de runtime
Esta seção contém informações sobre como diagnosticar e corrigir problemas de runtime com seu aplicativo Managed Service for Apache Flink.
**Topics**
+ [Ferramentas de solução de problemas](#troubleshooting-tools)
+ [Problemas do aplicativo](troubleshooting-symptoms.md)
+ [O aplicativo está sendo reiniciado](troubleshooting-rt-restarts.md)
+ [O throughput é muito lento](troubleshooting-rt-throughput.md)
+ [Crescimento ilimitado do estado](troubleshooting-rt-stateleaks.md)
+ [Operadores de E/S](troubleshooting-io-bound-operators.md)
+ [Controle de utilização de fonte ou upstream de um fluxo de dados do Kinesis](troubleshooting-source-throttling.md)
+ [Pontos de verificação](troubleshooting-checkpoints.md)
+ [O ponto de verificação está atingindo o tempo limite](troubleshooting-chk-timeout.md)
+ [Falha no ponto de verificação do aplicativo Apache Beam](troubleshooting-chk-failure-beam.md)
+ [Contrapressão](troubleshooting-backpressure.md)
+ [Distorção de dados](troubleshooting-data-skew.md)
+ [Distorção de estado](troubleshooting-state-skew.md)
+ [Integração com recursos em diferentes Regiões](troubleshooting-resources-in-different-regions.md)
## Ferramentas de solução de problemas
A principal ferramenta para detectar problemas de aplicativos são os CloudWatch alarmes. Usando CloudWatch alarmes, você pode definir limites para CloudWatch métricas que indicam condições de erro ou gargalo em seu aplicativo. Para obter informações sobre CloudWatch os alarmes recomendados, consulte[Use CloudWatch alarmes com o Amazon Managed Service para Apache Flink](monitoring-metrics-alarms.md).
# Problemas do aplicativo
Esta seção contém soluções para condições de erro que você pode encontrar com seu aplicativo Managed Service for Apache Flink.
**Topics**
+ [O aplicativo está preso em um status transitório](#troubleshooting-rt-stuck)
+ [Falha na criação do snapshot](#troubleshooting-rt-snapshots)
+ [Não é possível acessar recursos em uma VPC](#troubleshooting-rt-vpc)
+ [Dados são perdidos ao gravar em um bucket do Amazon S3](#troubleshooting-rt-s3)
+ [O aplicativo está no status EM EXECUÇÃO, mas não está processando dados](#troubleshooting-rt-processing)
+ [Erro de captura instantânea, atualização do aplicativo ou parada do aplicativo: InvalidApplicationConfigurationException](#troubleshooting-rt-appconfigexception)
+ [java.nio.file. NoSuchFileException:/usr/local/openjdk-8/lib/security/cacerts](#troubleshooting-rt-fnf)
## O aplicativo está preso em um status transitório
Se seu aplicativo permanecer em um status transitório (`STARTING`,`UPDATING`,`STOPPING`, ou`AUTOSCALING`), você poderá interrompê-lo usando a [StopApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_StopApplication.html)ação com o `Force` parâmetro definido como. `true` Você não pode forçar a interrupção de um aplicativo no status `DELETING`. Como alternativa, se o aplicativo estiver no status `UPDATING` ou `AUTOSCALING`, você poderá revertê-lo para a versão anterior em execução. Quando você reverte um aplicativo, ele carrega dados do estado do último snapshot bem-sucedido. Se o aplicativo não tiver snapshots, o Managed Service for Apache Flink rejeitará a solicitação de reversão. Para obter mais informações sobre como reverter um aplicativo, consulte [RollbackApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_RollbackApplication.html)ação.
**nota**
A interrupção forçada do aplicativo pode levar à perda ou duplicação de dados. Para evitar a perda de dados ou o processamento duplicado de dados durante a reinicialização do aplicativo, recomendamos que você tire instantâneos frequentes do seu aplicativo.
As causas para aplicativos travados incluem o seguinte:
+ **O estado do aplicativo é muito grande:** ter um estado de aplicativo muito grande ou muito persistente pode fazer com que o aplicativo fique preso durante uma operação de ponto de verificação ou instantâneo. Verifique as métricas `lastCheckpointDuration` e `lastCheckpointSize` do seu aplicativo para valores crescentes constantes ou valores anormalmente altos.
+ **O código do aplicativo é muito grande:** verifique se o arquivo JAR do aplicativo tem menos de 512 MB. Arquivos JAR maiores que 512 MB não são suportados.
+ **Falha na criação do instantâneo do aplicativo:** o Managed Service for Apache Flink tira um instantâneo do aplicativo durante uma solicitação de [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) ou [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StopApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_StopApplication.html). Em seguida, o serviço usa esse estado de instantâneo e restaura o aplicativo usando a configuração atualizada do aplicativo para fornecer uma semântica de processamento *exatamente uma vez*. Se a criação automática do instantâneo falhar, consulte [Falha na criação do snapshot](#troubleshooting-rt-snapshots) a seguir.
+ **Falha na restauração a partir de um instantâneo:** se você remover ou alterar um operador em uma atualização do aplicativo e tentar restaurar a partir de um instantâneo, a restauração falhará por padrão se o instantâneo contiver dados de estado do operador ausente. Além disso, o aplicativo ficará preso no status `STOPPED` ou `UPDATING`. Para alterar esse comportamento e permitir que a restauração seja bem-sucedida, altere o *AllowNonRestoredState*parâmetro do aplicativo [FlinkRunConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_FlinkRunConfiguration.html)para`true`. Isso permitirá que a operação de retomada ignore dados de estado que não possam ser mapeados para o novo programa.
+ **A inicialização do aplicativo está demorando mais:** o Managed Service for Apache Flink usa um tempo limite interno de 5 minutos (configuração suave) enquanto aguarda o início de uma tarefa do Flink. Se seu trabalho não começar dentro desse tempo limite, você verá um CloudWatch registro da seguinte forma:
```
Flink job did not start within a total timeout of 5 minutes for application: %s under account: %s
```
Se você encontrar o erro acima, isso significa que suas operações definidas no método `main` da tarefa do Flink estão demorando mais de 5 minutos, fazendo com que a criação da tarefa do Flink expire no final do Managed Service for Apache Flink. Sugerimos que você verifique os **JobManager**registros do Flink, bem como o código do aplicativo, para ver se esse atraso no `main` método é esperado. Caso contrário, você precisa adotar medidas para resolver o problema para que ele seja concluído em menos de 5 minutos.
Você pode verificar o status do seu aplicativo usando as ações [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html) ou [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html).
## Falha na criação do snapshot
O serviço Managed Service for Apache Flink não pode tirar um instantâneo nas seguintes circunstâncias:
+ O aplicativo excedeu o limite de instantâneos. O limite para instantâneos é de 1.000. Para obter mais informações, consulte [Gerenciar backups de aplicativos usando snapshots](how-snapshots.md).
+ O aplicativo não tem permissões para acessar sua fonte ou coletor.
+ O código do aplicativo não está funcionando corretamente.
+ O aplicativo está enfrentando outros problemas de configuração.
Se você receber uma exceção ao tirar um instantâneo durante uma atualização do aplicativo ou ao interromper o aplicativo, defina a propriedade `SnapshotsEnabled` de [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ApplicationSnapshotConfiguration.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ApplicationSnapshotConfiguration.html) do seu aplicativo como `false` e repita a solicitação.
Os instantâneos podem falhar se os operadores do seu aplicativo não forem provisionados adequadamente. Para obter informações sobre como ajustar o desempenho do operador, consulte [Escalonamento operador](performance-improving.md#performance-improving-scaling-op).
Depois que o aplicativo retornar ao estado íntegro, recomendamos que você defina a propriedade `SnapshotsEnabled` do aplicativo como `true`.
## Não é possível acessar recursos em uma VPC
Se seu aplicativo usa uma VPC em execução na Amazon VPC, faça o seguinte para verificar se seu aplicativo tem acesso aos recursos:
+ Verifique se há o seguinte erro em seus CloudWatch registros. Esse erro indica que seu aplicativo não pode acessar recursos em sua VPC:
```
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
```
Se você ver esse erro, verifique se suas tabelas de rotas estão configuradas corretamente e se seus conectores têm as configurações de conexão corretas.
Para obter informações sobre como configurar e analisar CloudWatch registros, consulte[Registro em log e monitoramento no Amazon Managed Service for Apache Flink](monitoring-overview.md).
## Dados são perdidos ao gravar em um bucket do Amazon S3
Pode ocorrer alguma perda de dados ao gravar a saída em um bucket do Amazon S3 usando o Apache Flink versão 1.6.2. Recomendamos usar a versão mais recente compatível do Apache Flink ao usar o Amazon S3 para saída direta. Para gravar em um bucket do Amazon S3 usando o Apache Flink 1.6.2, recomendamos o uso do Firehose. Para obter mais informações sobre como usar o Firehose com o Managed Service para Apache Flink, consulte [Coletor do Firehose](earlier.md#get-started-exercise-fh).
## O aplicativo está no status EM EXECUÇÃO, mas não está processando dados
Você pode verificar o status do seu aplicativo usando as ações [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_ListApplications.html) ou [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_DescribeApplication.html). Se seu aplicativo inserir o `RUNNING` status, mas não estiver gravando dados no coletor, você poderá solucionar o problema adicionando um stream de CloudWatch log da Amazon ao seu aplicativo. Para obter mais informações, consulte [Trabalhe com opções de CloudWatch registro de aplicativos](cloudwatch-logs.md#adding_cloudwatch). O fluxo de logs contém mensagens que podem ajudar a solucionar problemas do aplicativo.
## Erro de captura instantânea, atualização do aplicativo ou parada do aplicativo: InvalidApplicationConfigurationException
Um erro semelhante ao seguinte pode ocorrer durante uma operação de instantâneo ou durante uma operação que cria um instantâneo, como atualizar ou interromper um aplicativo:
```
An error occurred (InvalidApplicationConfigurationException) when calling the UpdateApplication operation:
Failed to take snapshot for the application xxxx at this moment. The application is currently experiencing downtime.
Please check the application's CloudWatch metrics or CloudWatch logs for any possible errors and retry the request.
You can also retry the request after disabling the snapshots in the Managed Service for Apache Flink console or by updating
the ApplicationSnapshotConfiguration through the AWS SDK
```
Esse erro ocorre quando o aplicativo não consegue criar um snapshot.
Se você encontrar esse erro durante uma operação de instantâneo ou uma operação que cria um instantâneo, faça o seguinte:
+ Desative os instantâneos do seu aplicativo. Você pode fazer isso no console do Managed Service for Apache Flink ou usando o `SnapshotsEnabledUpdate` parâmetro da [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html)ação.
+ Investigue por que os instantâneos não podem ser criados. Para obter mais informações, consulte [O aplicativo está preso em um status transitório](#troubleshooting-rt-stuck).
+ Reative os instantâneos quando o aplicativo retornar a um estado íntegro.
## java.nio.file. NoSuchFileException:/usr/local/openjdk-8/lib/security/cacerts
A localização do armazenamento confiável SSL foi atualizada em uma implantação anterior. Use o seguinte valor para o parâmetro `ssl.truststore.location`:
```
/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
```
# O aplicativo está sendo reiniciado
Se seu aplicativo não estiver íntegro, seu trabalho do Apache Flink falhará e será reiniciado continuamente. Esta seção descreve os sintomas e as etapas de solução de problemas dessa condição.
## Sintomas
Essa condição pode ter os seguintes sintomas:
+ A métrica `FullRestarts` não é zero. Essa métrica representa o número de vezes que o trabalho do aplicativo foi reiniciado desde que você iniciou o aplicativo.
+ A métrica `Downtime` não é zero. Essa métrica representa o número de milissegundos em que o aplicativo está no status `FAILING` ou `RESTARTING`.
+ O log do aplicativo contém alterações de status para `RESTARTING` ou `FAILED`. Você pode consultar o registro do seu aplicativo para essas mudanças de status usando a seguinte consulta do CloudWatch Logs Insights:[Analise os erros: falhas relacionadas à tarefa do aplicativo](cloudwatch-logs-reading.md#cloudwatch-logs-reading-apps).
## Causas e soluções
As condições a seguir podem fazer com que seu aplicativo fique instável e seja reiniciado repetidamente:
+ **O operador está lançando uma exceção:** se alguma exceção em um operador em seu aplicativo não for tratada, o aplicativo falhará (interpretando que a falha não pode ser tratada pelo operador). O aplicativo é reiniciado a partir do ponto de verificação mais recente para manter a semântica de processamento “exatamente uma vez”. Como resultado, o `Downtime` não é zero durante esses períodos de reinicialização. Para evitar que isso aconteça, recomendamos que você trate de quaisquer exceções que possam ser repetidas no código do aplicativo.
Você pode investigar as causas dessa condição consultando os logs do aplicativo em busca de alterações no estado do aplicativo de `RUNNING` para `FAILED`. Para obter mais informações, consulte [Analise os erros: falhas relacionadas à tarefa do aplicativo](cloudwatch-logs-reading.md#cloudwatch-logs-reading-apps).
+ **Os fluxos de dados do Kinesis não estão devidamente provisionados:** se uma fonte ou coletor do seu aplicativo for um fluxo de dados do Kinesis, verifique se há erros `ReadProvisionedThroughputExceeded` ou `WriteProvisionedThroughputExceeded` nas [métricas](https://docs.aws.amazon.com/streams/latest/dev/monitoring-with-cloudwatch.html) do fluxo.
Se você encontrar esses erros, poderá aumentar a throughput disponível para o fluxo do Kinesis aumentando o número de fragmentos do fluxo. Para obter mais informações, consulte [Como alterar o número de fragmentos abertos no Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-streams-open-shards/).
+ **Outras fontes ou coletores não estão adequadamente provisionados ou disponíveis:** verifique se seu aplicativo está provisionando corretamente as fontes e coletores. Verifique se todas as fontes ou coletores usados no aplicativo (como outros AWS serviços ou fontes ou destinos externos) estão bem provisionados, não estão sofrendo limitação de leitura ou gravação ou se estão periodicamente indisponíveis.
Se você estiver enfrentando problemas relacionados à throughput com seus serviços dependentes, aumente os recursos disponíveis para esses serviços ou investigue a causa de quaisquer erros ou indisponibilidade.
+ **Os operadores não são provisionados adequadamente:** se a workload nos threads de um dos operadores em seu aplicativo não for distribuída corretamente, o operador poderá ficar sobrecarregado e o aplicativo poderá falhar. Para obter informações sobre como ajustar o paralelismo do operador, consulte [Gerencie o escalonamento do operador adequadamente](performance-improving.md#performance-improving-scaling-op).
+ **O aplicativo falha com DaemonException:** Esse erro aparece no registro do aplicativo se você estiver usando uma versão do Apache Flink anterior à 1.11. Talvez seja necessário atualizar para uma versão posterior do Apache Flink para que uma versão KPL 0.14 ou posterior seja usada.
+ **O aplicativo falha com TimeoutException, FlinkException, ou RemoteTransportException:** Esses erros podem aparecer no registro do aplicativo se seus gerenciadores de tarefas estiverem falhando. Se seu aplicativo estiver sobrecarregado, seus gerenciadores de tarefas podem sofrer pressão de recursos de CPU ou memória, fazendo com que falhem.
Esses erros podem ter a seguinte aparência:
+ `java.util.concurrent.TimeoutException: The heartbeat of JobManager with id xxx timed out`
+ `org.apache.flink.util.FlinkException: The assigned slot xxx was removed`
+ `org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Connection unexpectedly closed by remote task manager`
Para solucionar essa condição, verifique o seguinte:
+ Verifique suas CloudWatch métricas quanto a picos incomuns no uso da CPU ou da memória.
+ Verifique se há problemas de throughput em seu aplicativo. Para obter mais informações, consulte [Solucione de problemas de desempenho](performance-troubleshooting.md).
+ Examine o log do aplicativo em busca de exceções não tratadas que o código do aplicativo está gerando.
+ **O aplicativo falha com o erro JaxbAnnotationModule Não encontrado:** esse erro ocorre se seu aplicativo usa o Apache Beam, mas não tem as dependências ou versões de dependências corretas. Os aplicativos do Managed Service for Apache Flink que usam o Apache Beam devem usar as seguintes versões de dependências:
```
2.10.2
...
com.fasterxml.jackson.module
jackson-module-jaxb-annotations
2.10.2
```
Se você não fornecer a versão correta do `jackson-module-jaxb-annotations` como uma dependência explícita, seu aplicativo a carrega das dependências do ambiente e, como as versões não coincidem, o aplicativo falha no runtime.
Para obter informações sobre como usar o Apache Beam com o Managed Service para Apache Flink, consulte. [Use CloudFormationCriar um aplicativo usando o Apache Beam](examples-beam.md)
+ **O aplicativo falha com java.io. IOException: Número insuficiente de buffers de rede**
Isso acontece quando um aplicativo não tem memória suficiente alocada para buffers de rede. Os buffers de rede facilitam a comunicação entre as subtarefas. Eles são usados para armazenar logs antes da transmissão por uma rede e para armazenar dados recebidos antes de dissecá-los em logs e entregá-los a subtarefas. O número de buffers de rede necessários é escalado diretamente com o paralelismo e a complexidade do seu gráfico de trabalho. Há várias abordagens para mitigar esse problema:
+ Você pode configurar um `parallelismPerKpu` menor para que haja mais memória alocada por subtarefa e buffers de rede. Observe que a redução de `parallelismPerKpu` aumentará a KPU e, portanto, o custo. Para evitar isso, você pode manter a mesma quantidade de KPU diminuindo o paralelismo pelo mesmo fator.
+ Você pode simplificar seu gráfico de tarefas reduzindo o número de operadores ou encadeando-os para que sejam necessários menos buffers.
+ Caso contrário, você pode entrar em contato https://aws.amazon.com/premiumsupport/ para obter uma configuração personalizada do buffer de rede.
# O throughput é muito lento
Se seu aplicativo não estiver processando os dados de transmissão recebidos com rapidez suficiente, ele terá um desempenho insatisfatório e ficará instável. Esta seção descreve os sintomas e as etapas de solução de problemas dessa condição.
## Sintomas
Essa condição pode ter os seguintes sintomas:
+ Se a fonte de dados do seu aplicativo for um fluxo do Kinesis, a métrica `millisbehindLatest` do fluxo aumentará continuamente.
+ Se a fonte de dados do seu aplicativo for um cluster do Amazon MSK, as métricas de atraso do consumidor do cluster aumentam continuamente. Para obter mais informações, consulte [Monitoramento de atraso do consumidor](https://docs.aws.amazon.com/msk/latest/developerguide/consumer-lag.html) no [Guia do desenvolvedor do Amazon MSK](https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html).
+ Se a fonte de dados do seu aplicativo for um serviço ou fonte diferente, verifique todas as métricas de atraso do consumidor ou dados disponíveis.
## Causas e soluções
Pode haver muitas causas para o throughput baixo do aplicativo. Se seu aplicativo não estiver acompanhando as entradas, verifique o seguinte:
+ Se o atraso no throughput estiver aumentando e depois diminuindo, verifique se o aplicativo está sendo reiniciado. Seu aplicativo interromperá o processamento da entrada enquanto for reiniciado, causando um aumento no atraso. Para obter informações sobre falhas do aplicativo, consulte [O aplicativo está sendo reiniciado](troubleshooting-rt-restarts.md).
+ Se o atraso no throughput for consistente, verifique se seu aplicativo está otimizado para desempenho. Para obter informações sobre como otimizar o desempenho do seu aplicativo, consulte [Solucione de problemas de desempenho](performance-troubleshooting.md).
+ Se o atraso no throughput não aumentar repentinamente, mas estiver aumentando continuamente, e seu aplicativo estiver otimizado para desempenho, você deverá aumentar os recursos do aplicativo. Para obter informações sobre o aumento dos recursos do aplicativo, consulte [Implemente a escalabilidade de aplicativos](how-scaling.md).
+ Se seu aplicativo lê de um cluster do Kafka em uma região diferente e `FlinkKafkaConsumer` ou `KafkaSource` está quase inativo (`idleTimeMsPerSecond` alto ou `CPUUtilization` baixo) apesar do alto atraso do consumidor, você pode aumentar o valor para `receive.buffer.byte`, como 2097152. Para obter mais informações, consulte a seção Ambiente de alta latência em [Configurações personalizadas do MSK](https://docs.aws.amazon.com/msk/latest/developerguide/msk-configuration-properties.html).
Para obter as etapas de solução de problemas relacionados ao throughput baixo ou ao aumento do atraso do consumidor na origem do aplicativo, consulte [Solucione de problemas de desempenho](performance-troubleshooting.md).
# Crescimento ilimitado do estado
Se seu aplicativo não estiver descartando adequadamente as informações de estado desatualizadas, elas se acumularão continuamente e causarão problemas de desempenho ou estabilidade do aplicativo. Esta seção descreve os sintomas e as etapas de solução de problemas dessa condição.
## Sintomas
Essa condição pode ter os seguintes sintomas:
+ A métrica `lastCheckpointDuration` está gradualmente aumentando ou apresentando picos.
+ A métrica `lastCheckpointSize` está gradualmente aumentando ou apresentando picos.
## Causas e soluções
As condições a seguir podem fazer com que seu aplicativo acumule dados de estado:
+ Seu aplicativo está retendo dados de estado por mais tempo do que o necessário.
+ Seu aplicativo usa consultas de janela com uma duração muito longa.
+ Você não definiu o TTL para os seus dados de estado. Para obter mais informações, consulte [State Time-To-Live (TTL) na documentação](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/datastream/fault-tolerance/state/#state-time-to-live-ttl) do Apache Flink.
+ Você está executando um aplicativo que depende da versão 2.25.0 ou mais recente do Apache Beam. Você pode optar por não participar da nova versão da transformação de leitura [ampliando sua BeamApplicationProperties](https://docs.aws.amazon.com/managed-flink/latest/java/examples-beam.html#examples-beam-configure) com os principais experimentos e valores`use_deprecated_read`. Para obter mais informações, consulte a [documentação do Apache Sqoop](https://beam.apache.org/blog/beam-2.25.0/#highlights).
Às vezes, os aplicativos enfrentam um crescimento cada vez maior do tamanho do estado, o que não é sustentável a longo prazo (afinal, um aplicativo Flink é executado indefinidamente). Às vezes, isso pode ser atribuído a aplicativos que armazenam dados no estado e não envelhecem adequadamente as informações antigas. Mas às vezes há expectativas irracionais sobre o que o Flink pode oferecer. Os aplicativos podem usar agregações em grandes janelas de tempo que abrangem dias ou até semanas. A menos que [AggregateFunctions](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/windows/#aggregatefunction)sejam usados, o que permite agregações incrementais, o Flink precisa manter os eventos de toda a janela no estado.
Além disso, ao usar funções de processo para implementar operadores personalizados, o aplicativo precisa remover dados de um estado que não é mais necessário para a lógica de negócios. Nesse caso, o [estado time-to-live](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/fault-tolerance/state/#state-time-to-live-ttl) pode ser usado para envelhecer automaticamente os dados com base no tempo de processamento. Managed Service for Apache Flink usa pontos de verificação incrementais e, portanto, o ttl de estado é baseado na[ compactação do RocksDB.](https://github.com/facebook/rocksdb/wiki/Compaction) Você só pode observar uma redução real no tamanho do estado (indicada pelo tamanho do ponto de verificação) após a ocorrência de uma operação de compactação. Em particular, para tamanhos de pontos de verificação abaixo de 200 MB, é improvável que você observe qualquer redução no tamanho dos pontos de verificação como resultado da expiração do estado. No entanto, os pontos de salvamento são baseados em uma cópia limpa do estado que não contém dados antigos, portanto, você pode acionar um instantâneo no Managed Service for Apache Flink para forçar a remoção do estado desatualizado.
Para fins de depuração, pode fazer sentido desativar os pontos de verificação incrementais para verificar mais rapidamente se o tamanho do ponto de verificação realmente diminui ou se estabiliza (e evitar o efeito da compactação no RockSBS). No entanto, isso requer um tíquete para a equipe de serviço.
# Operadores de E/S
É melhor evitar dependências de sistemas externos no caminho dos dados. Geralmente, é muito mais eficiente manter um conjunto de dados de referência em estado em vez de consultar um sistema externo para enriquecer eventos individuais. No entanto, às vezes há dependências que não podem ser facilmente transferidas para o estado, por exemplo, se você quiser enriquecer eventos com um modelo de machine learning hospedado no Amazon Sagemaker.
Os operadores que estão interagindo com sistemas externos pela rede podem se tornar um gargalo e causar contrapressão. É altamente recomendável usar o [AsyncIO](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/asyncio/) para implementar a funcionalidade, reduzir o tempo de espera de chamadas individuais e evitar que todo o aplicativo fique lento.
Além disso, para aplicativos com operadores I/O vinculados, também pode fazer sentido aumentar a configuração de [ParallelismPerKPU](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_ParallelismConfiguration.html) do aplicativo Managed Service for Apache Flink. Essa configuração descreve o número de subtarefas em paralelo que um aplicativo pode executar por unidade de processamento do Kinesis (KPU). Ao aumentar o valor do padrão de 1 para, digamos, 4, o aplicativo aproveita os mesmos recursos (e tem o mesmo custo), mas pode escalar até 4 vezes o paralelismo. Isso funciona bem para aplicativos I/O vinculados, mas causa sobrecarga adicional para aplicativos que não estão I/O vinculados.
# Controle de utilização de fonte ou upstream de um fluxo de dados do Kinesis
**Sintoma**: o aplicativo detecta `LimitExceededExceptions` a partir da fonte upstream do fluxo de dados do Kinesis.
**Causa provável**: a configuração padrão do conector Kinesis da biblioteca Apache Flink é definida para leitura da fonte do fluxo de dados do Kinesis com uma configuração padrão muito agressiva para o número máximo de registros buscados por chamada `GetRecords`. O Apache Flink é configurado por padrão para buscar 10.000 registros por chamada `GetRecords` (essa chamada é feita por padrão a cada 200 ms), embora o limite por fragmento seja de apenas 1.000 registros.
Esse comportamento padrão pode levar ao controle de utilização ao tentar consumir do fluxo de dados do Kinesis, o que afetará o desempenho e a estabilidade dos aplicativos.
Você pode confirmar isso verificando a CloudWatch `ReadProvisionedThroughputExceeded` métrica e vendo períodos prolongados ou sustentados em que essa métrica é maior que zero.
Você também pode ver isso nos CloudWatch registros do seu aplicativo Amazon Managed Service for Apache Flink observando erros contínuos. `LimitExceededException`
**Resolução**: você tem duas opções para resolver esse cenário:
+ Diminuir o limite padrão para o número de registros buscados por chamada de `GetRecords`
+ Habilitar as leituras adaptáveis em seu aplicativo Amazon Managed Service for Apache Flink. Para obter mais informações sobre o recurso Adaptive Reads, consulte [SHARD\$1USE\$1ADAPTIVE\$1READS](https://nightlies.apache.org/flink/flink-docs-release-1.10/api/java/org/apache/flink/streaming/connectors/kinesis/config/ConsumerConfigConstants.html#SHARD_USE_ADAPTIVE_READS)
# Pontos de verificação
Os pontos de verificação são o mecanismo do Flink para garantir que o estado de um aplicativo seja tolerante a falhas. O mecanismo permite que o Flink recupere o estado dos operadores se a tarefa falhar e fornece ao aplicativo a mesma semântica da execução sem falhas. Com o Managed Service for Apache Flink, o estado de um aplicativo é armazenado no RocksDB, um key/value armazenamento incorporado que mantém seu estado de funcionamento em disco. Quando um ponto de verificação é usado, o estado também é carregado no Amazon S3. Portanto, mesmo que o disco seja perdido, o ponto de verificação pode ser usado para restaurar o estado do aplicativo.
Para obter mais informações, consulte [Como o snapshot de estado funciona?](https://nightlies.apache.org/flink/flink-docs-master/docs/learn-flink/fault_tolerance/#how-does-state-snapshotting-work).
## Estágios do ponto de verificação
Para uma subtarefa de operador de ponto de verificação no Flink, há 5 estágios principais:
+ Aguardando [**Atraso inicial**]: o Flink usa barreiras de ponto de verificação que são inseridas no fluxo, então o tempo neste estágio é o tempo em que o operador espera que a barreira do ponto de verificação chegue até ela.
+ Alinhamento [**Duração do alinhamento**]: nesse estágio, a subtarefa atingiu uma barreira, mas está aguardando barreiras de outros fluxos de entrada.
+ Ponto de verificação de sincronização [**Duração da sincronização**]: esse estágio é quando a subtarefa realmente captura o estado do operador e bloqueia todas as outras atividades na subtarefa.
+ Ponto de verificação de assincronia [**Duração da assincronia**]: a maior parte desse estágio é a subtarefa de fazer o upload do estado para o Amazon S3. Durante esse estágio, a subtarefa não está mais bloqueada e pode processar registros.
+ Reconhecer — Geralmente é um estágio curto e é simplesmente a subtarefa de enviar uma confirmação para o JobManager e também realizar qualquer mensagem de confirmação (por exemplo, com coletores Kafka).
Cada um desses estágios (exceto Confirmação) é mapeado para uma métrica de duração para pontos de verificação que está disponível no Flink WebUI, o que pode ajudar a isolar a causa do longo ponto de verificação.
Para ver uma definição exata de cada uma das métricas disponíveis nos pontos de verificação, acesse a [guia Histórico](https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/ops/monitoring/checkpoint_monitoring/#history-tab).
## Investigar
Ao investigar a duração longa do ponto de verificação, a coisa mais importante a determinar é o gargalo do ponto de verificação, ou seja, qual operador e subtarefa estão demorando mais até o ponto de verificação e qual estágio dessa subtarefa está levando um período de tempo longo. Isso pode ser determinado usando o Flink WebUI na tarefa de ponto de verificação das tarefas. A interface web do Flink fornece dados e informações que ajudam a investigar problemas de ponto de verificação. Para obter uma análise completa, consulte [Monitoramento de pontos de verificação.](https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/ops/monitoring/checkpoint_monitoring/)
A primeira coisa a observar é a **duração de ponta a ponta** de cada operador no gráfico de tarefas para determinar qual operador está demorando até o ponto de verificação e merece uma investigação mais aprofundada. De acordo com a documentação do Flink, a definição da duração é:
*A duração do timestamp do acionador até a confirmação mais recente (ou n/a se nenhuma confirmação foi recebida ainda). Essa duração de ponta a ponta para um ponto de verificação completo é determinada pela última subtarefa que reconhece o ponto de verificação. Esse tempo geralmente é maior do que as subtarefas individuais necessárias para realmente verificar o estado.*
As outras durações do ponto de verificação também fornecem informações mais detalhadas sobre onde o tempo está sendo gasto.
Se a **Duração da sincronização** for alta, isso indica que algo está acontecendo durante o instantâneo. Durante esse estágio, `snapshotState()` é chamado para classes que implementam a interface SnapshotState; isso pode ser um código de usuário, portanto, os despejos de thread podem ser úteis para investigar isso.
Uma **Duração da assincronia** longa sugere que muito tempo está sendo gasto no upload do estado para o Amazon S3. Isso pode ocorrer se o estado for grande ou se houver muitos arquivos de estado sendo carregados. Se esse for o caso, vale a pena investigar como o estado está sendo usado pelo aplicativo e garantir que as estruturas de dados nativas do Flink estejam sendo usadas sempre que possível ([Uso do estado com chave](https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state/#using-keyed-state)). O Managed Service for Apache Flink configura o Flink de forma a minimizar o número de chamadas do Amazon S3 para garantir que não demore muito. A seguir há um exemplo das estatísticas de ponto de verificação de um operador. Isso mostra que a **Duração da assincronia** é relativamente longa em comparação com as estatísticas anteriores do ponto de verificação do operador.
![\[Investigar pontos de verificação\]](http://docs.aws.amazon.com/pt_br/managed-flink/latest/java/images/checkpoint.png)
O **Atraso inicial** alto mostra que a maior parte do tempo está sendo gasta esperando que a barreira do ponto de verificação chegue ao operador. Isso indica que o aplicativo está demorando um pouco para processar os registros, o que significa que a barreira está fluindo lentamente pelo gráfico de tarefas. Geralmente, esse é o caso se a tarefa estiver sob contrapressão ou se um ou mais operadores estiverem constantemente ocupados. Veja a seguir um exemplo de um JobGraph em que o segundo KeyedProcess operador está ocupado.
![\[Investigar pontos de verificação\]](http://docs.aws.amazon.com/pt_br/managed-flink/latest/java/images/checkpoint2.png)
Você pode investigar o que está demorando tanto usando Flink Flame Graphs ou TaskManager thread dumps. Uma vez identificado o gargalo, ele pode ser investigado mais detalhadamente usando o Flame graphs ou despejos de thread.
## Despejos de thread
Os despejos de thread são outra ferramenta de depuração que está em um nível um pouco mais baixo do que o Flame graphs. Um despejo de thread gera o estado de execução de todos os threads em um determinado momento. O Flink usa um despejo de thread JVM, que é um estado de execução de todos os threads no processo do Flink. O estado de um thread é apresentado por um rastreamento de pilha do thread, bem como por algumas informações adicionais. Na verdade, os Flame graphs são criados usando vários rastreamentos de pilha obtidos em rápida sucessão. O gráfico é uma visualização feita a partir desses traços que facilita a identificação dos caminhos de código comuns.
```
"KeyedProcess (1/3)#0" prio=5 Id=1423 RUNNABLE
at app//scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:154)
at $line33.$read$$iw$$iw$ExpensiveFunction.processElement(>19)
at $line33.$read$$iw$$iw$ExpensiveFunction.processElement(:14)
at app//org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
at app//org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:205)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
at app//org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
...
```
Acima está um trecho de um despejo de thread retirado da interface do usuário do Flink para um único thread. A primeira linha contém algumas informações gerais sobre esse thread, incluindo:
+ O nome do tópico *KeyedProcess (1/3) \$10*
+ Prioridade do thread *prio=5*
+ Uma ID de thread exclusiva *Id=1423*
+ Estado do thread *EXECUTÁVEL*
O nome de um thread geralmente fornece informações sobre o propósito geral do thread. Os encadeamentos do operador podem ser identificados pelo nome, pois os encadeamentos do operador têm o mesmo nome do operador, bem como uma indicação da subtarefa à qual estão relacionados, por exemplo, o encadeamento *KeyedProcess (1/3) \$10* é do *KeyedProcess*operador e da 1ª (de 3) subtarefa.
Os threads podem estar em um dos seguintes estados:
+ NOVO: o thread foi criado, mas ainda não foi processado
+ EXECUTÁVEL: o thread é executado na CPU
+ BLOQUEADO: o thread está esperando que outro thread libere seu bloqueio
+ AGUARDANDO: o thread está aguardando usando um método `wait()`, `join()` ou `park()`
+ TEMPORIZADO\$1AGUARDANDO: o thread está aguardando usando um método sleep, wait, join ou park, mas com um tempo de espera máximo.
**nota**
No Flink 1.13, a profundidade máxima de um único rastreamento de pilha no despejo de thread é limitada a 8.
**nota**
Os despejos de thread devem ser o último recurso para depurar problemas de desempenho em um aplicativo Flink, pois podem ser difíceis de ler e exigem que várias amostras sejam coletadas e analisadas manualmente. Se possível, é preferível usar Flame graphs.
### Despejos de thread no Flink
No Flink, um despejo de thread pode ser feito escolhendo a opção **Gerenciadores de tarefas** na barra de navegação esquerda da interface do usuário do Flink, selecionando um gerenciador de tarefas específico e navegando até a guia **Despejo de thread.** O despejo de thread pode ser baixado, copiado para seu editor de texto (ou analisador de thread dump) favorito ou analisado diretamente na visualização de texto na interface do usuário Web do Flink (no entanto, essa última opção pode ser um pouco complicada.
Para determinar qual gerenciador de tarefas usar, um despejo de tópicos da **TaskManagers**guia pode ser usado quando um determinado operador é escolhido. Isso mostra que o operador está sendo executado em diferentes subtarefas de um operador e pode ser executado em diferentes gerenciadores de tarefas.
![\[Uso de despejos de thread\]](http://docs.aws.amazon.com/pt_br/managed-flink/latest/java/images/checkpoint4.png)
O dump será composto por vários rastreamentos de pilha. No entanto, ao investigar o dump, os relacionados a um operador são os mais importantes. Eles podem ser facilmente encontrados, pois os threads do operador têm o mesmo nome do operador, bem como uma indicação da subtarefa à qual estão relacionados. Por exemplo, o rastreamento de pilha a seguir é do *KeyedProcess*operador e é a primeira subtarefa.
```
"KeyedProcess (1/3)#0" prio=5 Id=595 RUNNABLE
at app//scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:155)
at $line360.$read$$iw$$iw$ExpensiveFunction.processElement(:19)
at $line360.$read$$iw$$iw$ExpensiveFunction.processElement(:14)
at app//org.apache.flink.streaming.api.operators.KeyedProcessOperator.processElement(KeyedProcessOperator.java:83)
at app//org.apache.flink.streaming.runtime.tasks.OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask.java:205)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
at app//org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
at app//org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
...
```
Isso pode se tornar confuso se houver vários operadores com o mesmo nome, mas podemos nomear operadores para contornar isso. Por exemplo:
```
....
.process(new ExpensiveFunction).name("Expensive function")
```
## [Flame graphs](https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/ops/debugging/flame_graphs/)
Os Flame graphs são uma ferramenta de depuração útil que visualiza os rastreamentos da pilha do código de destino, o que permite identificar os caminhos de código mais frequentes. Eles são criados por meio de rastreamentos de pilha de amostras várias vezes. O eixo x de um Flame graph mostra os diferentes perfis da pilha, enquanto o eixo y mostra a profundidade da pilha e chama o rastreamento da pilha. Um único retângulo em um Flame graph representa uma estrutura de pilha, e a largura de uma chama mostra com que frequência ela aparece nas pilhas. Para obter mais detalhes sobre os Flame graphs e como usá-los, consulte [Flame graphs](https://www.brendangregg.com/flamegraphs.html).
No Flink, o gráfico de chama de um operador pode ser acessado por meio da interface do usuário da Web selecionando um operador e, em seguida, escolhendo a **FlameGraph**guia. Depois que amostras suficientes forem coletadas, o Flame graph será exibido. A seguir está a FlameGraph versão ProcessFunction que estava demorando muito para chegar ao posto de controle.
![\[Uso de Flame graphs\]](http://docs.aws.amazon.com/pt_br/managed-flink/latest/java/images/checkpoint3.png)
Este é um gráfico de chama muito simples e mostra que todo o tempo de CPU está sendo gasto em uma visão geral `processElement` do ExpensiveFunction operador. Você também obtém o número da linha para ajudar a determinar onde a execução do código está ocorrendo.
# O ponto de verificação está atingindo o tempo limite
Se seu aplicativo não for otimizado ou provisionado adequadamente, os pontos de verificação podem falhar. Esta seção descreve os sintomas e as etapas de solução de problemas dessa condição.
## Sintomas
Se os pontos de verificação falharem em seu aplicativo, o `numberOfFailedCheckpoints` será maior que zero.
Os pontos de verificação podem falhar devido a falhas diretas, como erros do aplicativo, ou devido a falhas transitórias, como a falta de recursos do aplicativo. Verifique os logs e as métricas do seu aplicativo para ver os seguintes sintomas:
+ Erros no seu código.
+ Erros ao acessar os serviços dependentes do seu aplicativo.
+ Erros ao serializar dados. Se o serializador padrão não conseguir serializar os dados do aplicativo, o aplicativo falhará. Para obter informações sobre como usar um serializador personalizado em seu aplicativo, consulte [Tipos de dados e serialização](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/dev/datastream/fault-tolerance/serialization/types_serialization/) na Documentação do Apache Flink.
+ Erros de falta de memória.
+ Picos ou aumentos constantes nas seguintes métricas:
+ `heapMemoryUtilization`
+ `oldGenerationGCTime`
+ `oldGenerationGCCount`
+ `lastCheckpointSize`
+ `lastCheckpointDuration`
Para obter mais informações sobre o monitoramento de pontos de verificação, consulte [Monitoramento de pontos de verificação](https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/ops/monitoring/checkpoint_monitoring/) na Documentação do Apache Flink.
## Causas e soluções
As mensagens de erro de log do aplicativo mostram a causa das falhas diretas. Falhas transitórias podem ter as seguintes causas:
+ Seu aplicativo tem provisionamento de KPU insuficiente. Para obter informações sobre como aumentar o provisionamento de aplicativos, consulte [Implemente a escalabilidade de aplicativos](how-scaling.md).
+ O tamanho do estado do seu aplicativo é muito grande. Você pode monitorar o tamanho do estado do seu aplicativo usando a métrica `lastCheckpointSize`.
+ Os dados de estado do seu aplicativo são distribuídos de forma desigual entre as chaves. Se seu aplicativo usa o operador `KeyBy`, verifique se os dados recebidos estão sendo divididos igualmente entre as chaves. Se a maioria dos dados estiver sendo atribuída a uma única chave, isso cria um gargalo que causa falhas.
+ Seu aplicativo está enfrentando contrapressão na memória ou na coleta de resíduos. Monitore `heapMemoryUtilization`, `oldGenerationGCTime` e `oldGenerationGCCount` do seu aplicativo em busca de picos ou valores cada vez maiores.
# Falha no ponto de verificação do aplicativo Apache Beam
Se seu aplicativo Beam estiver configurado com [shutdownSourcesAfterIdleMs](https://beam.apache.org/documentation/runners/flink/#:~:text=shutdownSourcesAfterIdleMs)0 ms, os pontos de verificação podem falhar ao serem acionados porque as tarefas estão no estado “CONCLUÍDO”. Esta seção descreve os sintomas e a resolução dessa condição.
## Sintomas
Acesse os CloudWatch registros do aplicativo Managed Service for Apache Flink e verifique se a seguinte mensagem de log foi registrada. A mensagem de log a seguir indica que o ponto de verificação falhou ao ser acionado, pois algumas tarefas foram concluídas.
```
{
"locationInformation": "org.apache.flink.runtime.checkpoint.CheckpointCoordinator.onTriggerFailure(CheckpointCoordinator.java:888)",
"logger": "org.apache.flink.runtime.checkpoint.CheckpointCoordinator",
"message": "Failed to trigger checkpoint for job your job ID since some tasks of job your job ID has been finished, abort the checkpoint Failure reason: Not all required tasks are currently running.",
"threadName": "Checkpoint Timer",
"applicationARN": your application ARN,
"applicationVersionId": "5",
"messageSchemaVersion": "1",
"messageType": "INFO"
}
```
Isso também pode ser encontrado no painel do Flink, onde algumas tarefas entraram no estado “CONCLUÍDO” e o ponto de verificação não é mais possível.
![\[Tarefas no estado “CONCLUÍDO”\]](http://docs.aws.amazon.com/pt_br/managed-flink/latest/java/images/beam_checkpoint_failure.png)
## Causa
shutdownSourcesAfterIdleMs é uma variável de configuração do Beam que desliga fontes que ficaram ociosas pelo tempo configurado de milissegundos. Depois que uma fonte é desligada, o ponto de verificação não é mais possível. Isso pode levar à [falha do ponto de verificação](https://issues.apache.org/jira/browse/FLINK-2491).
Uma das causas para as tarefas entrarem no estado “CONCLUÍDO” é quando shutdownSourcesAfter IdleMs está configurado para 0 ms, o que significa que as tarefas que estão ociosas serão encerradas imediatamente.
## Solução
Para evitar que as tarefas entrem imediatamente no estado “FINALIZADO”, defina como shutdownSourcesAfter IdleMs Long.MAX\$1VALUE. Isso pode ser feito de duas maneiras:
+ Opção 1: Se a configuração do beam estiver definida na página de configuração do aplicativo Managed Service for Apache Flink, você poderá adicionar um novo par de valores-chave para definir shutdpwnSourcesAfteridle Ms da seguinte forma:
![\[Definido como shutdownSourcesAfter IdleMs Long.MAX_VALUE\]](http://docs.aws.amazon.com/pt_br/managed-flink/latest/java/images/beam_checkpoint_failure_solution.png)
+ Opção 2: Se a configuração do feixe estiver definida no arquivo JAR, você poderá definir da shutdownSourcesAfter IdleMs seguinte forma:
```
FlinkPipelineOptions options = PipelineOptionsFactory.create().as(FlinkPipelineOptions.class); // Initialize Beam Options object
options.setShutdownSourcesAfterIdleMs(Long.MAX_VALUE); // set shutdownSourcesAfterIdleMs to Long.MAX_VALUE
options.setRunner(FlinkRunner.class);
Pipeline p = Pipeline.create(options); // attach specified options to Beam pipeline
```
# Contrapressão
O Flink usa contrapressão para adaptar a velocidade de processamento de operadores individuais.
O operador pode ter dificuldade em continuar processando o volume de mensagens que recebe por vários motivos. A operação pode exigir mais recursos de CPU do que o operador tem disponíveis. O operador pode esperar que I/O as operações sejam concluídas. Se um operador não consegue processar eventos com rapidez suficiente, isso gera contrapressão nos operadores a montante, alimentando o operador lento. Isso faz com que os operadores a montante diminuam a velocidade, o que pode propagar ainda mais a contrapressão para a fonte e fazer com que a fonte se adapte ao throughput geral do aplicativo, diminuindo também a velocidade. Você pode encontrar uma descrição mais detalhada da contrapressão e de como ela funciona em [Como o Apache Flink™ lida com a contrapressão](https://www.ververica.com/blog/how-flink-handles-backpressure).
Saber quais operadores em um aplicativo são lentos fornece informações cruciais para entender a causa raiz dos problemas de desempenho no aplicativo. As informações de contrapressão são [expostas por meio do painel do Flink](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/monitoring/back_pressure/). Para identificar o operador lento, procure o operador com um valor alto de contrapressão que esteja mais próximo de um coletor (operador B no exemplo a seguir). O operador que causa a lentidão é então um dos operadores a jusante (operador C no exemplo). O B pode processar eventos mais rapidamente, mas sofre contrapressão, pois não pode encaminhar a saída para o verdadeiro operador lento C.
```
A (backpressured 93%) -> B (backpressured 85%) -> C (backpressured 11%) -> D (backpressured 0%)
```
Depois de identificar o operador lento, tente entender por que ele é lento. Pode haver uma infinidade de motivos e, às vezes, não é óbvio o que está errado e pode exigir dias de depuração e criação de perfil para ser resolvido. A seguir estão alguns motivos óbvios e mais comuns, alguns dos quais são explicados mais detalhadamente abaixo:
+ O operador está fazendo E/S lentas, por exemplo, chamadas de rede (considere usar o AsynciO em vez disso).
+ Há uma distorção nos dados e um operador está recebendo mais eventos do que outros (verifique observando o número de mensagens in/out de subtarefas individuais (ou seja, instâncias do mesmo operador) no painel do Flink.
+ É uma operação que consome muitos recursos (se não houver distorção de dados, considere escalar para trabalho CPU/memory vinculado ou aumentar `ParallelismPerKPU` para trabalho I/O vinculado)
+ Registro de logs extensivo no operador (reduza o registro de logs ao mínimo para o aplicativo de produção ou considere enviar a saída de depuração para um fluxo de dados).
## Testar o throughput com o coletor de descarte
O [Coletor de descarte](https://nightlies.apache.org/flink/flink-docs-stable/api/java/org/apache/flink/streaming/api/functions/sink/DiscardingSink.html) simplesmente ignora todos os eventos que recebe enquanto ainda executa o aplicativo (um aplicativo sem nenhum coletor falha na execução). Isso é muito útil para testes de throughput, criação de perfis e para verificar se o aplicativo está escalando adequadamente. Também é uma verificação de integridade muito pragmática para verificar se os coletores estão causando contrapressão ou o aplicativo (mas verificar apenas as métricas de contrapressão geralmente é mais fácil e direto).
Ao substituir todos os coletores de um aplicativo por um coletor de descarte e criar uma fonte simulada que gera dados que se assemelham aos dados de produção, você pode medir o throughput máximo do aplicativo para uma determinada configuração de paralelismo. Em seguida, você também pode aumentar o paralelismo para verificar se o aplicativo está escalando adequadamente e não tem um gargalo que só surge com um throughput mais alto (por exemplo, devido à distorção de dados).
# Distorção de dados
Um aplicativo Flink é executado em um cluster de forma distribuída. Para aumentar a escala horizontalmente para vários nós, o Flink usa o conceito de fluxos com chave, o que significa essencialmente que os eventos de um fluxo são particionados de acordo com uma chave específica, por exemplo, ID do cliente, e o Flink pode então processar partições diferentes em nós diferentes. Muitos dos operadores do Flink então são avaliados com base nessas partições, por exemplo, [janelas com chave](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/windows/), [funções de processo](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/process_function/) e [E/S assíncrona](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/operators/asyncio/).
A escolha de uma chave de partição geralmente depende da lógica de negócios. Ao mesmo tempo, muitas das melhores práticas para, por exemplo, [DynamoDB](https://aws.amazon.com/dynamodb/) e Spark, se aplicam igualmente ao Flink, incluindo:
+ garantir uma alta cardinalidade das chaves de partição
+ evitar distorções no volume de eventos entre as partições
Você pode identificar distorções nas partições comparando os registros received/sent das subtarefas (ou seja, instâncias do mesmo operador) no painel do Flink. Além disso, o monitoramento do Managed Service for Apache Flink também pode ser configurado para expor métricas para `numRecordsIn/Out` e `numRecordsInPerSecond/OutPerSecond` em um nível de subtarefa.
# Distorção de estado
Para operadores com estado, ou seja, operadores que mantêm o estado de sua lógica de negócios, como janelas, a distorção de dados sempre leva à distorção de estado. Algumas subtarefas recebem mais eventos do que outras devido à distorção nos dados e, portanto, também persistem mais dados no estado. No entanto, mesmo para um aplicativo que tenha partições balanceadas uniformemente, pode haver uma distorção na quantidade de dados persistentes no estado. Por exemplo, para janelas de sessão, alguns usuários e sessões, respectivamente, podem ser muito mais longos do que outros. Se as sessões mais longas fizerem parte da mesma partição, isso pode levar a um desequilíbrio do tamanho do estado mantido por diferentes subtarefas do mesmo operador.
A distorção de estado não apenas aumenta mais recursos de memória e disco exigidos por subtarefas individuais, mas também pode diminuir o desempenho geral do aplicativo. Quando um aplicativo está passando por um ponto de verificação ou ponto de salvamento, o estado do operador persiste no Amazon S3, para proteger o estado contra falhas no nó ou no cluster. Durante esse processo (especialmente com exatamente uma semântica ativada por padrão no Managed Service para Apache Flink), o processamento é interrompido de uma perspectiva externa até que seja concluído. checkpoint/savepoint Se houver distorção de dados, o tempo para concluir a operação pode ser limitado por uma única subtarefa que tenha acumulado uma quantidade particularmente alta de estado. Em casos extremos, a checkpoints/savepoints execução pode falhar porque uma única subtarefa não consegue persistir no estado.
Assim como a distorção de dados, a distorção de estado pode reduzir substancialmente a velocidade de um aplicativo.
Para identificar a distorção de estado, você pode aproveitar o painel do Flink. Encontre um ponto de verificação ou ponto de salvamento recente e compare a quantidade de dados que foram armazenados para subtarefas individuais nos detalhes.
# Integração com recursos em diferentes Regiões
Você pode habilitar usando `StreamingFileSink` para gravar em um bucket do Amazon S3 em uma região diferente do seu aplicativo Managed Service for Apache Flink por meio de uma configuração necessária para replicação entre regiões na configuração do Flink. Para fazer isso, registre um ticket de suporte no [AWS Support Centro](https://console.aws.amazon.com/support/home#/).