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á.
# Gravar dados no Amazon Kinesis Data Streams
Um *produtor* é uma aplicação que grava dados no Amazon Kinesis Data Streams. Você pode criar produtores para o Kinesis Data Streams AWS SDK para Java usando a e a Kinesis Producer Library (KPL).
Se este é o primeiro contato com o Kinesis Data Streams, comece familiarizando-se com os conceitos e a terminologia apresentados em [O que é o Amazon Kinesis Data Streams?](introduction.md) e [Use o AWS CLI para realizar operações do Amazon Kinesis Data Streams](getting-started.md).
**Importante**
O Kinesis Data Streams oferece suporte a alterações do período de retenção do registro de dados no fluxo de dados. Para obter mais informações, consulte [Alterar o período de retenção de dados](kinesis-extended-retention.md).
Para colocar dados no fluxo, é necessário especificar o nome do fluxo, uma chave de partição e o blob de dados que serão adicionados ao fluxo. A chave de partição é usada para determinar em que fragmento do fluxo o registro de dados será adicionado.
Todos os dados no fragmento são enviados para o mesmo operador que está processando o fragmento. A chave de partição usada usa depende da lógica do aplicativo. O número de chaves de partição normalmente deve ser muito maior que o número de fragmentos. Isso ocorre porque a chave de partição é usada para determinar como mapear um registro de dados para um determinado fragmento. Se houver um número suficiente de chaves de partição, os dados podem ser distribuídos uniformemente pelos fragmentos de um fluxo.
**Topics**
+ [Desenvolver produtores usando a Amazon Kinesis Producer Library (KPL)](developing-producers-with-kpl.md)
+ [Desenvolva produtores usando a API Amazon Kinesis Data Streams com o AWS SDK para Java](developing-producers-with-sdk.md)
+ [Gravar no Amazon Kinesis Data Streams usando o Kinesis Agent](writing-with-agents.md)
+ [Grave no Kinesis Data Streams usando outros serviços AWS](using-other-services.md)
+ [Gravar no Kinesis Data Streams usando integrações de terceiros](using-other-services-third-party.md)
+ [Solução de problemas de produtores do Amazon Kinesis Data Streams](troubleshooting-producers.md)
+ [Otimizar os produtores do Kinesis Data Streams](advanced-producers.md)
# Desenvolver produtores usando a Amazon Kinesis Producer Library (KPL)
Um produtor do Amazon Kinesis Data Streams é uma aplicação que coloca registros de dados de usuários em um fluxo de dados do Kinesis (o que também é chamado de *ingestão de dados*). A Amazon Kinesis Producer Library (KPL) simplifica o desenvolvimento de aplicações de produção, permitindo que os desenvolvedores atinjam alto throughput de gravação para um fluxo de dados do Kinesis.
Você pode monitorar o KPL com a Amazon CloudWatch. Para obter mais informações, consulte [Monitore a Kinesis Producer Library com a Amazon CloudWatch](monitoring-with-kpl.md).
**Topics**
+ [Analisar a função da KPL](#developing-producers-with-kpl-role)
+ [Perceber as vantagens de usar a KPL](#developing-producers-with-kpl-advantage)
+ [Entender quando não usar a KPL](#developing-producers-with-kpl-when)
+ [Instalar a KPL](kinesis-kpl-dl-install.md)
+ [Migrar da KPL 0.x para a KPL 1.x](kpl-migration-1x.md)
+ [Transição para certificados Amazon Trust Services (ATS) para a KPL](kinesis-kpl-upgrades.md)
+ [Plataformas compatíveis com a KPL](kinesis-kpl-supported-plats.md)
+ [Principais conceitos da KPL](kinesis-kpl-concepts.md)
+ [Integrar a KPL com o código de produtor](kinesis-kpl-integration.md)
+ [Gravar no fluxo de dados do Kinesis usando a KPL](kinesis-kpl-writing.md)
+ [Configurar a Amazon Kinesis Producer Library](kinesis-kpl-config.md)
+ [Implementar a desagregação de consumidores](kinesis-kpl-consumer-deaggregation.md)
+ [Usar a KPL com o Amazon Data Firehose](kpl-with-firehose.md)
+ [Use o KPL com o Registro do AWS Glue Esquema](kpl-with-schemaregistry.md)
+ [Definir a configuração do proxy da KPL](kpl-proxy-configuration.md)
+ [Política de ciclo de vida da versão da KPL](kpl-version-lifecycle-policy.md)
**nota**
Recomendamos a atualização para a versão mais recente da KPL. A KPL é atualizada regularmente em versões que incluem os patches de dependência e segurança e as correções de bugs mais recentes, além de novos recursos compatíveis com versões anteriores. Para obter mais informações, consulte [https://github.com/awslabs/amazon-kinesis-producer/releases/](https://github.com/awslabs/amazon-kinesis-producer/releases/).
## Analisar a função da KPL
A KPL é uma easy-to-use biblioteca altamente configurável que ajuda você a gravar em um stream de dados do Kinesis. Ela atua como um intermediário entre o código da aplicação de produção e as ações das APIs do Kinesis Data Streams. A KPL executa as seguintes tarefas principais:
+ Grava em um ou mais fluxos de dados do Kinesis com um mecanismo automático e configurável de novas tentativas
+ Coleta registros e usa `PutRecords` para gravar vários registros em vários fragmentos por solicitação
+ Agrega registros de usuário para aumentar o tamanho da carga útil e melhorar o throughput
+ Integra-se perfeitamente à [Kinesis Client Library](https://docs.aws.amazon.com/kinesis/latest/dev/developing-consumers-with-kcl.html) (KCL) para desagregar registros em lote no consumidor
+ Envia CloudWatch métricas da Amazon em seu nome para fornecer visibilidade sobre o desempenho do produtor
Observe que o KPL é diferente da API Kinesis Data Streams que está disponível no. [AWS SDKs](https://aws.amazon.com/tools/) A API do Kinesis Data Streams ajuda a gerenciar vários aspectos do Kinesis Data Streams (incluindo a criação de fluxos, a refragmentação e a colocação e obtenção de registros), enquanto a KPL fornece uma camada de abstração especificamente para a ingestão de dados. Para obter informações sobre a API do Kinesis Data Streams, consulte a [Referência de APIs do Amazon Kinesis](https://docs.aws.amazon.com/kinesis/latest/APIReference/).
## Perceber as vantagens de usar a KPL
A lista a seguir apresenta algumas das principais vantagens no uso da KPL para desenvolver produtores do Kinesis Data Streams.
A KPL pode ser usada em casos de uso síncronos ou assíncronos. Recomenda-se a interface assíncrona, que possui maior desempenho, a menos que haja um motivo específico para usar o comportamento síncrono. Para obter mais informações sobre esses dois casos de uso e o código de exemplo, consulte [Gravar no fluxo de dados do Kinesis usando a KPL](kinesis-kpl-writing.md).
**Benefícios de desempenho**
A KPL pode ajudar a criar produtores de alta performance. Considere uma situação em que suas instâncias do Amazon EC2 atuam como proxy para coletar eventos de 100 bytes de centenas ou milhares de dispositivos de baixa potência e gravar registros em um fluxo de dados do Kinesis. Cada instância do EC2 precisa gravar milhares de eventos por segundo no fluxo de dados. Para alcançar o throughput necessário, os produtores precisam implementar uma lógica complexa, como agrupamento em lotes ou multithreading, lógica de retentativa e desagregação de registros no lado do consumidor. A KPL executa todas essas tarefas.
**Facilidade de uso do lado do consumidor**
No caso de desenvolvedores no lado do consumidor usando a KCL em Java, a integração da KPL não requer esforço adicional. Ao recuperar um registro agregado do Kinesis Data Streams que consiste em vários registros de usuário da KPL, a KCL chama automaticamente a KPL para extrair registros do usuário individual e retorná-los ao mesmo usuário.
Desenvolvedores no lado do conusmidor que não usam a KCL, mas usam a operação `GetRecords` da API diretamente, contam com uma biblioteca de Java da KPL disponível para extrair registros do usuário individual e retorná-los ao mesmo usuário.
**Monitoramento de produtor**
Você pode coletar, monitorar e analisar seus produtores do Kinesis Data Streams usando a CloudWatch Amazon e a KPL. O KPL emite taxa de transferência, erro e outras métricas CloudWatch em seu nome e é configurável para ser monitorado no nível de stream, fragmento ou produtor.
**Arquitetura assíncrona**
Como pode armazenar registros em buffer antes de enviá-los ao Kinesis Data Streams, a KPL não força a aplicação chamadora a bloquear e aguardar confirmação de que o registro chegou ao servidor antes de continuar o runtime. Uma chamada para colocar um registro na KPL sempre retorna imediatamente, sem esperar o registro ser enviado ou uma resposta ser recebida do servidor. Em vez disso, é criado um objeto `Future` que posteriormente recebe o resultado do envio do registro ao Kinesis Data Streams. Esse é o mesmo comportamento dos clientes assíncronos no SDK. AWS
## Entender quando não usar a KPL
A KPL pode ter um atraso por processamento adicional de até `RecordMaxBufferedTime` dentro da biblioteca (configurável pelo usuário). Valores maiores que `RecordMaxBufferedTime` geram maiores eficiências de empacotamento e melhor desempenho. Aplicações que não toleram esse atraso adicional podem precisar usar o SDK da AWS diretamente. Para obter mais informações sobre como usar o AWS SDK com o Kinesis Data Streams, consulte. [Desenvolva produtores usando a API Amazon Kinesis Data Streams com o AWS SDK para Java](developing-producers-with-sdk.md) Para obter mais informações sobre `RecordMaxBufferedTime` e outras propriedades configuráveis pelo usuário da KPL, consulte [Configurar a Amazon Kinesis Producer Library](kinesis-kpl-config.md).
# Instalar a KPL
A Amazon fornece binários pré-compilados da Amazon Kinesis Producer Library (KPL) de C\$1\$1 para macOS, Windows e distribuições recentes do Linux (para obter detalhes das plataformas com suporte, consulte a próxima seção). Esses binários, empacotados como parte dos arquivos .jar do Java, são invocados e usados automaticamente ao usar o Maven para instalar o pacote. Para localizar as versões mais recentes da KPL e da KCL, use os seguintes links de pesquisa do Maven:
+ [KPL](https://search.maven.org/#search|ga|1|amazon-kinesis-producer)
+ [KCL](https://search.maven.org/#search|ga|1|amazon-kinesis-client)
O binários do Linux foram compilados com GNU Compiler Collection (GCC) e vinculados estaticamente à libstdc\$1\$1 no Linux. Espera-se que eles funcionem em qualquer distribuição do Linux de 64 bits que inclua um glibc versão 2.5 ou superior.
Usuários de distribuições Linux anteriores podem criar o KPL usando as instruções de compilação fornecidas junto com o código-fonte ativado. GitHub Para baixar o KPL de GitHub, consulte a [Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-producer) Producer Library.
**Importante**
A Amazon Kinesis Producer Library (KPL) 0.x chegará em 30 de end-of-support janeiro de 2026. É **altamente recomendável** que você migre suas aplicações da KPL que usam a versão 0.x para a versão mais recente da KPL antes de 30 de janeiro de 2026. Para saber a versão mais recente da KPL, consulte a [página da KPL no Github](https://github.com/awslabs/amazon-kinesis-producer). Para ter mais informações sobre como migrar da KPL 0.x para a KPL 1.x, consulte [Migrar da KPL 0.x para a KPL 1.x](kpl-migration-1x.md).
# Migrar da KPL 0.x para a KPL 1.x
Este tópico fornece step-by-step instruções para migrar seu consumidor do KPL 0.x para o KPL 1.x. O KPL 1.x introduz suporte para o AWS SDK para Java 2.x, mantendo a compatibilidade da interface com as versões anteriores. Não é necessário atualizar sua lógica de processamento dos dados principais para migrar para a KPL 1.x.
1. **É necessário atender aos seguintes pré-requisitos:**
+ Java Development Kit (JDK) 8 ou posterior
+ AWS SDK para Java 2. x
+ Maven ou Gradle para gerenciamento de dependências
1. **Adicionar dependências**
Se estiver usando Maven, adicione a dependência a seguir ao arquivo pom.xml. Atualize groupid de `com.amazonaws` para `software.amazon.kinesis` e a versão `1.x.x` para a versão mais recente da KPL.
```
software.amazon.kinesis
amazon-kinesis-producer
1.x.x
```
Se você estiver usando Gradle, adicione o seguinte ao seu arquivo `build.gradle`. Substitua `1.x.x` pela versão mais recente da KPL.
```
implementation 'software.amazon.kinesis:amazon-kinesis-producer:1.x.x'
```
A versão mais recente da KPL pode ser obtida no [Repositório central do Maven](https://central.sonatype.com/search?q=amazon-kinesis-producer).
1. **Atualizar as instruções de importação para a KPL**
O KPL 1.x usa o AWS SDK para Java 2.x e usa um nome de pacote atualizado que começa com`software.amazon.kinesis`, em comparação com o nome do pacote no KPL anterior que começa com. `com.amazonaws.services.kinesis`
Substitua a importação de `com.amazonaws.services.kinesis` por `software.amazon.kinesis`. A tabela a seguir relaciona as importações que precisam ser substituídas.
**Substituições de importações**
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/pt_br/streams/latest/dev/kpl-migration-1x.html)
1. **Atualizar declarações de importação para classes de provedores AWS de credenciais**
Ao migrar para o KPL 1.x, você deve atualizar pacotes e classes em suas importações no código do aplicativo KPL que são baseados no AWS SDK para Java 1.x para os correspondentes com base no 2.x. AWS SDK para Java As importações comuns na aplicação KPL são classes de provedores de credenciais. Consulte [Alterações no provedor de credenciais](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) na documentação do guia de migração AWS SDK para Java 2.x para ver a lista completa de alterações no provedor de credenciais. Aqui está a alteração comum de importação que talvez você precise fazer em suas aplicações KPL.
**Importação na KPL 0.x**
```
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
```
**Importação na KPL 1.x**
```
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
```
Se você importar outros provedores de credenciais com base no AWS SDK para Java 1.x, deverá atualizá-los para os equivalentes ao AWS SDK para Java 2.x. Se você não importou nada classes/packages do AWS SDK para Java 1.x, pode ignorar essa etapa.
1. **Atualize a configuração do provedor de credenciais na configuração da KPL**
A configuração do provedor de credenciais na KPL 1.x requer os provedores de credenciais do AWS SDK para Java 2.x. Se você estiver passando provedores de credenciais para o AWS SDK para Java 1.x no `KinesisProducerConfiguration` substituindo o provedor de credenciais padrão, você deverá atualizá-lo com os AWS SDK para Java provedores de credenciais 2.x. Consulte [Alterações no provedor de credenciais](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html) na documentação do guia de migração AWS SDK para Java 2.x para ver a lista completa de alterações no provedor de credenciais. Se não substituiu o provedor de credenciais padrão na configuração da KPL, você pode ignorar essa etapa.
Por exemplo, se você estiver substituindo o provedor de credenciais padrão da KPL com o seguinte código:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// SDK v1 default credentials provider
config.setCredentialsProvider(new DefaultAWSCredentialsProviderChain());
```
É necessário atualizá-lo com o código a seguir para usar o provedor de credenciais do AWS SDK para Java 2.x:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration();
// New SDK v2 default credentials provider
config.setCredentialsProvider(DefaultCredentialsProvider.create());
```
# Transição para certificados Amazon Trust Services (ATS) para a KPL
Em 9 de fevereiro de 2018, às 9:00, horário do Pacífico, o Amazon Kinesis Data Streams instalou os certificados do ATS. Para poder continuar a gravar registros no Kinesis Data Streams usando a Amazon Kinesis Producer Library (KPL), é necessário atualizar a instalação da KPL para a [versão 0.12.6](http://search.maven.org/#artifactdetails|com.amazonaws|amazon-kinesis-producer|0.12.6|jar) ou posterior. Essa mudança afeta todas as AWS regiões.
Para obter informações sobre a mudança para o ATS, consulte [Como se preparar para AWS a mudança para sua própria autoridade de certificação](https://aws.amazon.com/blogs/security/how-to-prepare-for-aws-move-to-its-own-certificate-authority/).
Se encontrar problemas e precisar de suporte técnico, [abra um caso](https://console.aws.amazon.com/support/v1#/case/create) na central de suporte da AWS .
# Plataformas compatíveis com a KPL
A Amazon Kinesis Producer Library (KPL) é escrita em C\$1\$1 e executada como um processo secundário do processo principal do usuário. Binários nativos pré-compilados de 64 bits são fornecidos com a versão do Java e gerenciados pelo wrapper Java.
O pacote Java é executado sem a necessidade de instalar qualquer biblioteca adicional nos seguintes sistemas operacionais:
+ Distribuições do Linux com kernel 2.6.18 (setembro de 2006) e posterior
+ Apple iOS X 10.9 e posterior
+ Windows Server 2008 e posterior
**Importante**
O Windows Server 2008 e posterior é compatível com todas as versões do KPL até a versão 0.14.0.
A plataforma Windows NÃO é compatível a partir da versão KPL 0.14.0 ou superior.
Observe que a KPL é de 64 bits apenas.
## Código-fonte
Se os binários fornecidos na instalação da KPL não forem suficientes para seu ambiente, o núcleo da KPL é escrito como um módulo C\$1\$1. O código-fonte do módulo C\$1\$1 e da interface Java é lançado sob a Amazon Public License e está disponível GitHub na [Amazon Kinesis](https://github.com/awslabs/amazon-kinesis-producer) Producer Library. Embora seja possível usar a KPL em qualquer plataforma para a qual haja disponibilidade de JRE e um compilador C\$1\$1 compatível com padrões recentes, a Amazon não oferece suporte oficial a nenhuma plataforma que não esteja na lista de plataformas compatíveis.
# Principais conceitos da KPL
As seções a seguir contêm os conceitos e a terminologia necessários para entender e aproveitar a Amazon Kinesis Producer Library (KPL).
**Topics**
+ [Registros](#kinesis-kpl-concepts-records)
+ [Agrupamento em lotes](#kinesis-kpl-concepts-batching)
+ [Agregação](#kinesis-kpl-concepts-aggretation)
+ [Coleta](#kinesis-kpl-concepts-collection)
## Registros
Neste guia, há uma distinção entre *registros de usuário da KPL* e *registros do Kinesis Data Streams*. O termo *registro* sem um qualificador refere-se a um *registro de usuário da KPL*. Um registro do Kinesis Data Streams será explicitamente chamado de *registro do Kinesis Data Streams*.
Um registro de usuário da KPL é um blob de dados com um significado específico para o usuário. Os exemplos incluem um blob JSON que representa um evento de interface do usuário em um site ou uma entrada de log proveniente de um servidor da web.
Um registro do Kinesis Data Streams é uma instância da estrutura de dados de `Record` definida pela API do serviço Kinesis Data Streams. Ele contém uma chave de partição, um número sequencial e um blob de dados.
## Agrupamento em lotes
*Envio em lotes* refere-se à execução de uma única ação em vários itens, em vez de executar a ação repetidamente em cada item.
Nesse contexto, o “item” é um registro, e a ação é seu envio ao Kinesis Data Streams. Em uma situação sem envio em lotes, cada registro é colocado em um registro separado do Kinesis Data Streams e faz uma única solicitação HTTP para enviá-lo ao Kinesis Data Streams. Com o agrupamento em lotes, cada solicitação HTTP pode carregar vários registros, em vez de apenas um.
A KPL aceita dois tipos de agrupamento em lotes:
+ *Agregação*: armazenamento de vários registros em um único registro do Kinesis Data Streams.
+ *Coleta*: uso da operação `PutRecords` da API para enviar vários registros do Kinesis Data Streams a um ou mais fragmentos no fluxo de dados do Kinesis.
Os dois tipos de agrupamento em lotes da KPL são projetados para coexistir e podem ser ativados ou desativados de forma independente. Por padrão, ambos são ativados.
## Agregação
A *agregação* refere-se ao armazenamento de vários registros em um registro do Kinesis Data Streams. A agregação permite que os clientes aumentem o número de registros enviados por chamada de API, o que aumenta efetivamente o throughput do produtor.
Os fragmentos do Kinesis Data Streams oferecem suporte a até 1.000 registros do Kinesis Data Streams por segundo, ou um throughput de 1 MB. O limite de registros por segundo do Kinesis Data Streams restringe clientes com registros menores que 1 KB. A agregação de registros permite aos clientes combinar vários registros em um único registro do Kinesis Data Streams. Isso permite que os clientes melhorem o próprio throughput por fragmento.
Considere o caso de um fragmento na região us-east-1 que está atualmente em execução à taxa constante de 1.000 registros por segundo, com registros de 512 bytes cada. Com a agregação da KPL, é possível empacotar 1.000 registros em apenas 10 registros do Kinesis Data Streams reduzindo o número de registros por segundo para 10 (a 50 KB cada).
## Coleta
*Coleta* refere-se ao agrupamento de vários registros do Kinesis Data Streams em um lote, que é enviado em uma única solicitação HTTP com uma chamada à operação `PutRecords` da API, em vez de enviar cada registro do Kinesis Data Streams em uma solicitação HTTP própria.
Isso aumenta o throughput em comparação com a não utilização de coleções, pois reduz a sobrecarga de fazer muitas solicitações HTTP separadas. Na verdade, `PutRecords`, por si só, foi projetado especificamente para essa finalidade.
A coleta difere da agregação porque opera com grupos de registros do Kinesis Data Streams. Os registros do Kinesis Data Streams coletados ainda podem conter vários registros do usuário. O relacionamento pode ser visualizado da seguinte maneira:
```
record 0 --|
record 1 | [ Aggregation ]
... |--> Amazon Kinesis record 0 --|
... | |
record A --| |
|
... ... |
|
record K --| |
record L | | [ Collection ]
... |--> Amazon Kinesis record C --|--> PutRecords Request
... | |
record S --| |
|
... ... |
|
record AA--| |
record BB | |
... |--> Amazon Kinesis record M --|
... |
record ZZ--|
```
# Integrar a KPL com o código de produtor
A Amazon Kinesis Producer Library (KPL) é executada em um processo separado e se comunica com o processo principal do usuário usando IPC. Essa arquitetura, às vezes chamada de [microsserviço](http://en.wikipedia.org/wiki/Microservices), é escolhida por dois motivos principais:
**1) O processo do usuário não falhará mesmo que a KPL falhe**
O processo pode ter tarefas não relacionadas ao Kinesis Data Streams e continuar em operação mesmo que a KPL falhe. Além disso, o processo de usuário pai pode reiniciar a KPL e recuperar um estado totalmente funcional (essa funcionalidade está nos wrappers oficiais).
Um exemplo é um servidor Web que envia métricas ao Kinesis Data Streams. O servidor pode continuar entregando páginas mesmo que a parte do Kinesis Data Streams tenha parado de funcionar. Portanto, causar uma falha em todo o servidor devido a um erro na KPL seria uma interrupção desnecessária.
**2) Clientes arbitrários podem ser aceitos**
Há sempre clientes que usam linguagens diferentes das oficialmente aceitas. Esses clientes também devem ser capazes de usar a KPL facilmente.
## Matriz de uso recomendado
A matriz de uso a seguir lista as configurações recomendadas para diferentes usuários e indica se e como a KPL deve ser usada. Lembre-se de que, se a agregação estiver habilitada, será preciso usar a desagregação para extrair seus registros no lado do consumidor.
| Linguagem do lado do produtor | Linguagem do lado do consumidor | Versão da KCL | Lógica do ponto de verificação | A KPL pode ser usada? | Advertências |
| --- | --- | --- | --- | --- | --- |
| Tudo menos Java | \$1 | \$1 | \$1 | Não | N/D |
| Java | Java | Usa o Java SDK diretamente | N/D | Sim | Se a agregação for usada, será necessário usar a biblioteca de desagregação fornecida após as chamadas a GetRecords. |
| Java | Tudo menos Java | Usa o SDK diretamente | N/D | Sim | É preciso desabilitar a agregação. |
| Java | Java | 1.3.x | N/D | Sim | É preciso desabilitar a agregação. |
| Java | Java | 1.4.x | Chama o ponto de verificação sem nenhum argumento | Sim | Nenhum |
| Java | Java | 1.4.x | Chama o ponto de verificação com um número sequencial explícito | Sim | Desative a agregação ou altere o código para usar números sequenciais estendidos para definir pontos de verificação. |
| Java | Tudo menos Java | 1.3.x \$1 daemon de várias linguagens \$1 wrapper específico de linguagem | N/D | Sim | É preciso desabilitar a agregação. |
# Gravar no fluxo de dados do Kinesis usando a KPL
As seções a seguir mostram o código de exemplo em uma progressão desde o produtor mais básico até o código totalmente assíncrono.
## Código barebone de produtor
O código a seguir é todo o necessário para escrever um produtor minimamente funcional. Os registros de usuário da Amazon Kinesis Producer Library (KPL) são processados em segundo plano.
```
// KinesisProducer gets credentials automatically like
// DefaultAWSCredentialsProviderChain.
// It also gets region automatically from the EC2 metadata service.
KinesisProducer kinesis = new KinesisProducer();
// Put some records
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
kinesis.addUserRecord("myStream", "myPartitionKey", data);
}
// Do other stuff ...
```
## Responder aos resultados de forma síncrona
No exemplo anterior, o código não verificou se os registros do usuário da KPL foram bem-sucedidos. A KPL faz todas as novas tentativas necessárias para dar conta das falhas. No entanto, para verificar os resultados, é possível utilizar os objetos `Future` que são retornados de `addUserRecord`, como no exemplo a seguir (exemplo anterior mostrado para fins de contexto):
```
KinesisProducer kinesis = new KinesisProducer();
// Put some records and save the Futures
List> putFutures = new LinkedList>();
for (int i = 0; i < 100; i++) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
// doesn't block
putFutures.add(
kinesis.addUserRecord("myStream", "myPartitionKey", data));
}
// Wait for puts to finish and check the results
for (Future f : putFutures) {
UserRecordResult result = f.get(); // this does block
if (result.isSuccessful()) {
System.out.println("Put record into shard " +
result.getShardId());
} else {
for (Attempt attempt : result.getAttempts()) {
// Analyze and respond to the failure
}
}
}
```
## Responder aos resultados de forma assíncrona
O exemplo anterior está chamando `get()` em um objeto `Future`, o que bloqueia o runtime. Se não quiser bloquear o runtime, você pode usar callback assíncrono, como mostrado no exemplo a seguir:
```
KinesisProducer kinesis = new KinesisProducer();
FutureCallback myCallback = new FutureCallback() {
@Override public void onFailure(Throwable t) {
/* Analyze and respond to the failure */
};
@Override public void onSuccess(UserRecordResult result) {
/* Respond to the success */
};
};
for (int i = 0; i < 100; ++i) {
ByteBuffer data = ByteBuffer.wrap("myData".getBytes("UTF-8"));
ListenableFuture f = kinesis.addUserRecord("myStream", "myPartitionKey", data);
// If the Future is complete by the time we call addCallback, the callback will be invoked immediately.
Futures.addCallback(f, myCallback);
}
```
# Configurar a Amazon Kinesis Producer Library
Embora as configurações padrão devam funcionar bem para a maioria dos casos de uso, talvez convenha alterar algumas configurações padrão para ajustar o comportamento do `KinesisProducer` às suas necessidades. Para isso, uma instância da classe `KinesisProducerConfiguration` pode ser passada ao construtor `KinesisProducer`, por exemplo:
```
KinesisProducerConfiguration config = new KinesisProducerConfiguration()
.setRecordMaxBufferedTime(3000)
.setMaxConnections(1)
.setRequestTimeout(60000)
.setRegion("us-west-1");
final KinesisProducer kinesisProducer = new KinesisProducer(config);
```
Também é possível pode carregar uma configuração de um arquivo de propriedades:
```
KinesisProducerConfiguration config = KinesisProducerConfiguration.fromPropertiesFile("default_config.properties");
```
É possível substituir qualquer caminho e nome de arquivo a que o processo de usuário tem acesso. Também é possível chamar métodos definidos para a instância de `KinesisProducerConfiguration` criada dessa forma para personalizar a configuração.
O arquivo de propriedades deve especificar parâmetros usando seus nomes em PascalCase. Os nomes correspondem aos usados nos métodos definidos na classe `KinesisProducerConfiguration`. Por exemplo:
```
RecordMaxBufferedTime = 100
MaxConnections = 4
RequestTimeout = 6000
Region = us-west-1
```
Para obter mais informações sobre regras de uso de parâmetros de configuração e limites de valor, consulte o [arquivo de exemplo de propriedades de configuração em GitHub](https://github.com/awslabs/amazon-kinesis-producer/blob/master/java/amazon-kinesis-producer-sample/default_config.properties).
Observe que, depois que o `KinesisProducer` é inicializado, alterar a instância de `KinesisProducerConfiguration` que foi usada não tem mais efeito. No momento, o `KinesisProducer` não oferece suporte à reconfiguração dinâmica.
# Implementar a desagregação de consumidores
A partir da versão 1.4.0, a KCL oferece suporte à desagregação automática dos registros de usuário da KPL. O código da aplicação de consumo escrito com versões anteriores da KCL será compilado sem qualquer modificação após a atualziação da KCL. No entanto, se a agregação da KPL estiver sendo usada no lado do produtor, haverá uma sutileza envolvendo a definição de pontos de verificação: como todos os sub-registros dentro de um registro agregado têm o mesmo número de sequência, os dados adicionais precisarão ser armazenados com o ponto de verificação, se for preciso distinguir os sub-registros. Esses dados adicionais são chamados de *números de subsequência*.
**Topics**
+ [Migrar de versões anteriores da KCL](#kinesis-kpl-consumer-deaggregation-migration)
+ [Usar extensões da KCL para desagregação da KPL](#kinesis-kpl-consumer-deaggregation-extensions)
+ [Use GetRecords diretamente](#kinesis-kpl-consumer-deaggregation-getrecords)
## Migrar de versões anteriores da KCL
Não é necessário alterar as chamadas existentes para que definam pontos de verificação com a agregação. A recuperação bem-sucedida de todos os registros armazenados no Kinesis Data Streams ainda é garantida. A KCL agora oferece duas novas operações de ponto de verificação para suporte a casos de uso específicos, descritos abaixo.
Caso seu código existente tenha sido escrito para uma versão da KCL anterior ao suporte à KPL e a operação de ponto de verificação seja chamada sem argumentos, isso será equivalente a definir o ponto de verificação do número de sequência do último registro de usuário da KPL no lote. Se a operação de ponto de verificação é chamada com uma string de número sequencial, isso equivale a definir o ponto de verificação do número sequencial do lote conhecido junto com o número 0 (zero) da subsequência implícita.
Chamar a nova operação de ponto de verificação `checkpoint()` da KCL sem argumentos é semanticamente equivalente a definir o ponto de verificação do número de sequência da última chamada a `Record` no lote com o número 0 (zero) de subsequência implícito.
Chamar a nova operação de ponto de verificação `checkpoint(Record record)` da KCL é semanticamente equivalente a definir o ponto de verificação do número de sequência do `Record` conhecido com o número 0 (zero) de subsequência implícito. Se a chamada a `Record` é na verdade um `UserRecord`, o número de sequencial e subsequencial de `UserRecord` será definido.
Chamar a nova operação de ponto de verificação `checkpoint(String sequenceNumber, long subSequenceNumber)` da KCL explicitamente define o ponto de verificação do número de sequência conhecido com o número de subsequência conhecido.
Em qualquer desses casos, depois que o ponto de verificação é armazenado na tabela de pontos de verificação do Amazon DynamoDB, a KCL pode retomar corretamente a recuperação de registros, mesmo quando a aplicação falha e reinicia. Se houver mais registros contidos na sequência, a recuperação ocorrerá a partir do próximo registro de número subsequencial dentro do registro com o número sequencial cujo ponto de verificação foi definido mais recentemente. Se o ponto de verificação mais recente inclui o último número subsequencial do registro de número sequencial anterior, a recuperação ocorrerá a partir do registro com o próximo número sequencial.
A próxima seção discute detalhes de definição do ponto de verificação de sequência e subsequência para consumidores que precisam evitar a omissão e a duplicação de registros. Se a omissão (ou duplicação) de registros ao parar e reiniciar o processamento de registros do consumidor não é importante, é possível executar seu código existente sem modificação.
## Usar extensões da KCL para desagregação da KPL
A desagregação da KPL pode envolver a definição de um ponto de verificação de subsequência. Para facilitar a definição do ponto de verificação de subsequência, uma classe `UserRecord` foi adicionada à KCL:
```
public class UserRecord extends Record {
public long getSubSequenceNumber() {
/* ... */
}
@Override
public int hashCode() {
/* contract-satisfying implementation */
}
@Override
public boolean equals(Object obj) {
/* contract-satisfying implementation */
}
}
```
Essa classe agora é usada em vez de `Record`. Ela não interrompe o código existente por ser uma subclasse de `Record`. A classe `UserRecord` representa os sub-registros reais e os registros padrão não agregados. Os registros não agregados podem ser considerados como registros agregadas com exatamente um sub-registro.
Além disso, duas operações novas são adicionadas a `IRecordProcessorCheckpointer`:
```
public void checkpoint(Record record);
public void checkpoint(String sequenceNumber, long subSequenceNumber);
```
Para começar a usar a definição de ponto de verificação de número subsequencial, é possível executar a seguinte conversão. Altere o seguinte código de formulário:
```
checkpointer.checkpoint(record.getSequenceNumber());
```
Novo código de formulário:
```
checkpointer.checkpoint(record);
```
Recomendamos usar o formulário `checkpoint(Record record)` para definição de ponto de verificação de subsequência. No entanto, se `sequenceNumbers` já estiverem sendo armazenados em strings para usar na definição de pontos de verificação, agora deve-se também armazenar `subSequenceNumber`, como mostrado no exemplo a seguir:
```
String sequenceNumber = record.getSequenceNumber();
long subSequenceNumber = ((UserRecord) record).getSubSequenceNumber(); // ... do other processing
checkpointer.checkpoint(sequenceNumber, subSequenceNumber);
```
A transmissão de `Record` para `UserRecord` sempre é bem-sucedida porque a implementação sempre usa `UserRecord`. A menos que haja uma necessidade de executar aritmética nos números sequenciais, essa abordagem não é recomendada.
Durante o processamento de registros de usuário da KPL, a KCL grava o número de subsequência no Amazon DynamoDB como um campo extra para cada linha. As versões anteriores da KCL usavam `AFTER_SEQUENCE_NUMBER` para obter registros ao retomar os pontos de verificação. Em vez disso, a KCL atual com suporte à KPL, usa `AT_SEQUENCE_NUMBER`. Quando é recuperado o registro no número sequencial para o qual foi definido o ponto de verificação, esse número é verificado e os sub-registros são descartados conforme apropriado (que podem ser todos eles, se o último sub-registro é aquele verificado). Novamente, os registros não agregados podem ser considerados como registros agregados com um único sub-registro. Portanto, o mesmo algoritmo funciona para os registros agregados e não agregados.
## Use GetRecords diretamente
Em vez de usar a KCL, pode-se optar por invocar a operação de API `GetRecords` diretamente para recuperar os registros do Kinesis Data Streams. Para desempacotar os registros recuperados nos registros de usuário originais da KPL, chame uma das seguintes operações estáticas em `UserRecord.java`:
```
public static List deaggregate(List records)
public static List deaggregate(List records, BigInteger startingHashKey, BigInteger endingHashKey)
```
A primeira operação usa o valor `0` (zero) padrão para `startingHashKey` e o valor `2^128 -1` padrão para `endingHashKey`.
Cada uma dessas operações desagrega a lista de registros conhecidos do Kinesis Data Streams em uma lista de registros de usuário da KPL. Qualquer registro de usuário da KPL com uma chave de hash explícita ou chave de partição fora do intervalo de `startingHashKey` (inclusive) e de `endingHashKey` (inclusive) é descartado da lista de registros retornados.
# Usar a KPL com o Amazon Data Firehose
Ao usar a Kinesis Producer Library (KPL) para gravar dados em um fluxo de dados do Kinesis, é possível usar agregação para combinar os registros gravados. Ao usar esse fluxo de dados como origem para o fluxo de entrega do Firehose, O Firehose desagregará os registros antes de entregá-los ao destino. Quando o fluxo de entrega é configurado para transformar os dados, o Firehose desagregará os registros antes de entregá-los ao AWS Lambda. Para obter mais informações, consulte [Gravando no Amazon Firehose Usando o Kinesis Data Streams](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-kinesis-streams.html).
# Use o KPL com o Registro do AWS Glue Esquema
Você pode integrar seus streams de dados do Kinesis com o AWS Glue Schema Registry. O registro de esquemas do AWS Glue permite detectar, controlar e evoluir esquemas centralmente, ao mesmo tempo que garante que os dados produzidos sejam validados continuamente por um esquema registrado. O esquema define a estrutura e o formato de um registro de dados. Um esquema é uma especificação versionada para publicação, consumo ou datastore confiáveis. O AWS Glue Schema Registry permite que você melhore a qualidade end-to-end dos dados e a governança de dados em seus aplicativos de streaming. Para obter mais informações, consulte [Registro de esquemas do AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html). Uma das formas de configurar essa integração é usar as bibliotecas KPL e Kinesis Client Library (KCL) em Java.
**Importante**
Atualmente, a integração do Kinesis Data AWS Glue Streams e do registro de esquemas só é compatível com os streams de dados do Kinesis que usam produtores de KPL implementados em Java. Não há suporte para múltiplas linguagens.
Para obter instruções detalhadas sobre como configurar a integração do Kinesis Data Streams com o Schema Registry usando o KPL, consulte a [seção “Interagindo com dados KPL/KCL usando as bibliotecas” em Caso de uso: integração do Amazon Kinesis Data Streams](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) com o Glue Schema Registry. AWS
# Definir a configuração do proxy da KPL
Para aplicativos que não podem se conectar diretamente à Internet, todos os clientes do AWS SDK oferecem suporte ao uso de proxies HTTP ou HTTPS. Em um ambiente empresarial típico, todo o tráfego de saída da rede precisa passar por servidores proxy. Se seu aplicativo usa a Kinesis Producer Library (KPL) para coletar e enviar dados AWS em um ambiente que usa servidores proxy, seu aplicativo exigirá a configuração do proxy KPL. A KPL é uma biblioteca de alto nível criada com base no SDK do AWS Kinesis. Ele é dividido em um processo nativo e um wrapper. O processo nativo executa todas as tarefas de processamento e envio de registros, enquanto o wrapper gerencia o processo nativo e se comunica com ele. Para obter mais informações, consulte [Implementar Produtores Eficientes e Confiáveis com a Amazon Kinesis Producer Library](https://aws.amazon.com/blogs/big-data/implementing-efficient-and-reliable-producers-with-the-amazon-kinesis-producer-library/).
O wrapper é escrito em Java e o processo nativo é escrito em C\$1\$1 com o uso do SDK do Kinesis. A KPL versão 0.14.7 ou superior é compatível com a configuração de proxy no wrapper Java, que pode passar todas as configurações de proxy para o processo nativo. Para obter mais informações, consulte [https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v0.14.7](https://github.com/awslabs/amazon-kinesis-producer/releases/tag/v0.14.7).
É possível usar o código a seguir para adicionar configurações de proxy às aplicações da KPL.
```
KinesisProducerConfiguration configuration = new KinesisProducerConfiguration();
// Next 4 lines used to configure proxy
configuration.setProxyHost("10.0.0.0"); // required
configuration.setProxyPort(3128); // default port is set to 443
configuration.setProxyUserName("username"); // no default
configuration.setProxyPassword("password"); // no default
KinesisProducer kinesisProducer = new KinesisProducer(configuration);
```
# Política de ciclo de vida da versão da KPL
Este tópico descreve a política de ciclo de vida da versão para a Amazon Kinesis Producer Library (KPL). AWS fornece regularmente novos lançamentos para as versões do KPL para oferecer suporte a novos recursos e aprimoramentos, correções de erros, patches de segurança e atualizações de dependências. Recomendamos que você continue up-to-date com as versões do KPL para acompanhar os recursos, as atualizações de segurança e as dependências subjacentes mais recentes. **Não** recomendamos o uso contínuo de uma versão não compatível da KPL.
O ciclo de vida das principais versões da KPL consiste nas três fases a seguir:
+ **Disponibilidade geral (GA)** — Durante essa fase, a versão principal é totalmente suportada. AWS fornece lançamentos regulares de versões secundárias e de patches que incluem suporte para novos recursos ou atualizações de API para o Kinesis Data Streams, bem como correções de bugs e segurança.
+ **Modo de manutenção** — AWS limita os lançamentos da versão do patch para tratar apenas de correções críticas de bugs e problemas de segurança. A versão principal não receberá atualizações dos novos recursos ou APIs do Kinesis Data Streams.
+ **E nd-of-support** — A versão principal não receberá mais atualizações ou lançamentos. As versões publicadas anteriormente continuarão disponíveis por meio de gerenciadores de pacotes públicos e o código permanecerá ativado GitHub. O uso de uma versão que chegou end-of-support é feito a critério do usuário. Recomendamos que você faça a atualização para a versão principal mais recente.
| Versão principal | Fase atual | Data de lançamento | Data do modo de manutenção | End-of-support encontro |
| --- | --- | --- | --- | --- |
| KPL 0.x | Modo de manutenção | 2015-06-02 | 2025-04-17 | 2026-01-30 |
| KPL 1.x | Disponibilidade geral | 2024-12-15 | -- | -- |
# Desenvolva produtores usando a API Amazon Kinesis Data Streams com o AWS SDK para Java
Você pode desenvolver produtores usando a API Amazon Kinesis Data Streams AWS com o SDK for Java. Se você nunca usou o Kinesis Data Streams, comece familiarizando-se com os conceitos e a terminologia apresentados em [O que é o Amazon Kinesis Data Streams?](introduction.md) e [Use o AWS CLI para realizar operações do Amazon Kinesis Data Streams](getting-started.md).
Esses exemplos discutem a [API do Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/APIReference/) e usam o [AWS SDK para Java](https://aws.amazon.com/sdk-for-java/) para adicionar (colocar) dados a um fluxo. Contudo, na maioria dos casos de uso, é melhor usar a biblioteca KPL do Kinesis Data Streams. Para obter mais informações, consulte [Desenvolver produtores usando a Amazon Kinesis Producer Library (KPL)](developing-producers-with-kpl.md).
O código Java de exemplo neste capítulo demonstra como executar operações básicas da API do Kinesis Data Streams e está dividido logicamente por tipo de operação. Esses exemplos não representam um código pronto para produção, pois não verificam todas as exceções possíveis nem abrangem todas as considerações de segurança ou de performance possíveis. Também é possível chamar a [API do Kinesis Data Streams](https://docs.aws.amazon.com/kinesis/latest/APIReference/) usando outras linguagens de programação. Para obter mais informações sobre todas as opções disponíveis AWS SDKs, consulte [Comece a desenvolver com a Amazon Web Services](https://aws.amazon.com/developers/getting-started/).
Cada tarefa tem pré-requisitos. Por exemplo, não é possível adicionar dados a um fluxo enquanto o fluxo não é criado, o que requer a criação de um cliente. Para obter mais informações, consulte [Criar e gerenciar fluxos de dados do Kinesis](working-with-streams.md).
**Topics**
+ [Adicionar dados a um fluxo](#kinesis-using-sdk-java-add-data-to-stream)
+ [Interaja com os dados usando o AWS Glue Schema Registry](kinesis-integration-glue-schema-registry.md)
## Adicionar dados a um fluxo
Quando um fluxo é criado, pode-se adicionar dados a ele na forma de registros. Um registro é uma estrutura de dados que contém os dados a serem processados na forma de um blob de dados. Depois de armazenar os dados no registro, o Kinesis Data Streams não inspeciona, interpreta nem altera dados de forma alguma. Cada registro também tem um número sequencial e uma chave de partição associados.
Há duas operações diferentes na API do Kinesis Data Streams que adicionam dados a um fluxo, [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) e [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html). A operação `PutRecords` envia vários registros ao fluxo por solicitação HTTP e a operação singular `PutRecord` envia registros ao fluxo um por vez (uma solicitação HTTP separada é necessária para cada registro). Talvez convenha usar `PutRecords` para a maioria dos aplicativos, pois ele atingirá um throughput mais alto por produtor de dados. Para obter mais informações sobre cada uma dessas operações, consulte as subseções abaixo.
**Topics**
+ [Adicione vários registros com PutRecords](#kinesis-using-sdk-java-putrecords)
+ [Adicione um único registro com PutRecord](#kinesis-using-sdk-java-putrecord)
Sempre tenha em mente que, como a aplicação de origem está adicionando dados ao fluxo usando a API do Kinesis Data Streams, provavelmente há uma ou mais aplicações de consumo processando dados fora do fluxo simultaneamente. Para obter informações sobre como os consumidores obtêm dados usando a API do Kinesis Data Streams, consulte [Como obter dados de um fluxo](developing-consumers-with-sdk.md#kinesis-using-sdk-java-get-data).
**Importante**
[Alterar o período de retenção de dados](kinesis-extended-retention.md)
### Adicione vários registros com PutRecords
A operação [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) envia vários registros ao Kinesis Data Streams em uma única solicitação. Ao usar `PutRecords`, os produtores podem obter um throughput mais alto ao enviar dados para o fluxo de dados do Kinesis. Cada solicitação `PutRecords` pode oferecer suporte a até 500 registros. Cada registro na solicitação pode ter no máximo 1 MB, até um limite de 5 MB para toda a solicitação, incluindo chaves de partição. Assim como a operação única `PutRecord` descrita abaixo, `PutRecords` usa números de sequência e chaves de partição. No entanto, o parâmetro `PutRecord` de `SequenceNumberForOrdering` não é incluído em uma chamada a `PutRecords`. A operação `PutRecords` tenta processar todos os registros na ordem natural da solicitação.
Cada registro de dados tem um número sequencial exclusivo. O número de sequência é atribuído pelo Kinesis Data Streams depois que `client.putRecords` é chamada para adicionar os registros de dados ao fluxo. Os números sequenciais da mesma chave de partição geralmente aumentam com o tempo: quanto maior o período entre as solicitações `PutRecords`, maiores ficam os números sequenciais.
**nota**
Os números de sequência não podem ser usados como índices para conjuntos de dados dentro do mesmo fluxo. Para separar logicamente conjuntos de dados, use chaves de partição ou crie um fluxo separado para cada conjunto de dados.
Uma solicitação `PutRecords` pode incluir registros com diferentes chaves de partição. O escopo da solicitação é um fluxo; cada solicitação pode incluir qualquer combinação de chaves de partição e registros, dentro dos limites da solicitação. As solicitações feitas com diferentes chaves de partição a streams com muitos fragmentos diferentes costumam ser mais rápidas do que as solicitações com um pequeno número de chaves de partição para um pequeno número de fragmentos. O número de chaves de partição deve ser muito maior do que o número de fragmentos para reduzir a latência e maximizar o throughput.
#### PutRecords exemplo
O código a seguir cria 100 registros de dados com chaves de partição sequenciais e os coloca em um fluxo denominado `DataStream`.
```
AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
clientBuilder.setRegion(regionName);
clientBuilder.setCredentials(credentialsProvider);
clientBuilder.setClientConfiguration(config);
AmazonKinesis kinesisClient = clientBuilder.build();
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(streamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int i = 0; i < 100; i++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(i).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", i));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = kinesisClient.putRecords(putRecordsRequest);
System.out.println("Put Result" + putRecordsResult);
```
A resposta a `PutRecords` inclui uma matriz de resposta `Records`. Cada registro na matriz de resposta se correlaciona diretamente com um registro na matriz de solicitação por ordenação natural, do início ao fim da solicitação e da resposta. A matriz de resposta de `Records` sempre inclui o mesmo número de registros da matriz de solicitação.
#### Lidar com falhas ao usar PutRecords
Por padrão, a falha de registros individuais em uma solicitação não interrompe o processamento de registros subsequentes em uma solicitação `PutRecords`. Isso significa que uma matriz de resposta `Records` inclui os registros com processamento bem-sucedido e malsucedido. É preciso detectar os registros com processamento malsucedido e incluí-los em uma chamada subsequente.
Os registros bem-sucedidos incluem os valores `SequenceNumber` e `ShardID`, e os registros malsucedidos incluem os valores `ErrorCode` e `ErrorMessage`. O parâmetro `ErrorCode` reflete o tipo de erro e pode ter um dos seguintes valores: `ProvisionedThroughputExceededException` ou `InternalFailure`. `ErrorMessage` fornece informações mais detalhadas sobre a exceção `ProvisionedThroughputExceededException`, incluindo o ID da conta, o nome do fluxo e o ID do fragmento do registro que foi limitado. O exemplo abaixo tem três registros em uma solicitação `PutRecords`. O segundo registro falha e isso é refletido na resposta.
**Example PutRecords Sintaxe da solicitação**
```
{
"Records": [
{
"Data": "XzxkYXRhPl8w",
"PartitionKey": "partitionKey1"
},
{
"Data": "AbceddeRFfg12asd",
"PartitionKey": "partitionKey1"
},
{
"Data": "KFpcd98*7nd1",
"PartitionKey": "partitionKey3"
}
],
"StreamName": "myStream"
}
```
**Example PutRecords Sintaxe de resposta**
```
{
"FailedRecordCount”: 1,
"Records": [
{
"SequenceNumber": "21269319989900637946712965403778482371",
"ShardId": "shardId-000000000001"
},
{
“ErrorCode":”ProvisionedThroughputExceededException”,
“ErrorMessage": "Rate exceeded for shard shardId-000000000001 in stream exampleStreamName under account 111111111111."
},
{
"SequenceNumber": "21269319989999637946712965403778482985",
"ShardId": "shardId-000000000002"
}
]
}
```
Os registros com processamento malsucedido podem ser incluídos nas solicitações `PutRecords` subsequentes. Primeiro, verifique o parâmetro `FailedRecordCount` no `putRecordsResult` para confirmar se há registros com falha na solicitação. Assim sendo, cada `putRecordsEntry` com um `ErrorCode` que não seja `null` deve ser adicionado a uma solicitação subsequente. Para obter um exemplo desse tipo de handler, consulte o seguinte código.
**Example PutRecords manipulador de falhas**
```
PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
putRecordsRequest.setStreamName(myStreamName);
List putRecordsRequestEntryList = new ArrayList<>();
for (int j = 0; j < 100; j++) {
PutRecordsRequestEntry putRecordsRequestEntry = new PutRecordsRequestEntry();
putRecordsRequestEntry.setData(ByteBuffer.wrap(String.valueOf(j).getBytes()));
putRecordsRequestEntry.setPartitionKey(String.format("partitionKey-%d", j));
putRecordsRequestEntryList.add(putRecordsRequestEntry);
}
putRecordsRequest.setRecords(putRecordsRequestEntryList);
PutRecordsResult putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
while (putRecordsResult.getFailedRecordCount() > 0) {
final List failedRecordsList = new ArrayList<>();
final List putRecordsResultEntryList = putRecordsResult.getRecords();
for (int i = 0; i < putRecordsResultEntryList.size(); i++) {
final PutRecordsRequestEntry putRecordRequestEntry = putRecordsRequestEntryList.get(i);
final PutRecordsResultEntry putRecordsResultEntry = putRecordsResultEntryList.get(i);
if (putRecordsResultEntry.getErrorCode() != null) {
failedRecordsList.add(putRecordRequestEntry);
}
}
putRecordsRequestEntryList = failedRecordsList;
putRecordsRequest.setRecords(putRecordsRequestEntryList);
putRecordsResult = amazonKinesisClient.putRecords(putRecordsRequest);
}
```
### Adicione um único registro com PutRecord
Cada chamada para [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html) opera em um único registro. Prefira a operação `PutRecords` descrita em [Adicione vários registros com PutRecords](#kinesis-using-sdk-java-putrecords), a menos que seu aplicativo precise especificamente enviar sempre registos únicos por solicitação ou algum outro motivo para o não uso de `PutRecords`.
Cada registro de dados tem um número sequencial exclusivo. O número de sequência é atribuído pelo Kinesis Data Streams depois que `client.putRecord` é chamada para adicionar o registro de dados ao fluxo. Os números sequenciais da mesma chave de partição geralmente aumentam com o tempo: quanto maior o período entre as solicitações `PutRecord`, maiores ficam os números sequenciais.
Quando ocorrem colocações em rápida sucessão, não há garantia de que os números de sequência retornados aumentem, porque as operações put aparentam ser essencialmente simultâneas para o Kinesis Data Streams. Para garantir estritamente o aumento de números sequenciais para a mesma chave de partição, use o parâmetro `SequenceNumberForOrdering`, como mostrado no código de exemplo em [PutRecord exemplo](#kinesis-using-sdk-java-putrecord-example).
Usando ou não `SequenceNumberForOrdering`, os registros que o Kinesis Data Streams recebe por meio de uma chamada a `GetRecords` são estritamente ordenados por número de sequência.
**nota**
Os números de sequência não podem ser usados como índices para conjuntos de dados dentro do mesmo fluxo. Para separar logicamente conjuntos de dados, use chaves de partição ou crie um fluxo separado para cada conjunto de dados.
Uma chave de partição é usada para agrupar os dados dentro de um fluxo. Um registro de dados é atribuído a um fragmento dentro do fluxo com base em sua chave de partição. Especificamente, o Kinesis Data Streams usa a chave de partição como entrada para uma função de hash que mapeia essa chave (e os dados associados) a um determinado fragmento.
Como resultado desse mecanismo de hashing, todos os registros de dados com a mesma chave de partição são mapeados para o mesmo fragmento no fluxo. No entanto, se o número de chaves de partição ultrapassar o número de fragmentos, alguns fragmentos conterão necessariamente registros com chaves de partição diferentes. Do ponto de vista do design, para garantir que todos os seus fragmentos sejam bem utilizados, o número de fragmentos (especificado pelo método `setShardCount` de `CreateStreamRequest`) deve ser substancialmente menor que o número de chaves de partição exclusivas, e o volume de dados que flui para uma única chave de partição deve ser substancialmente menor que a capacidade do fragmento.
#### PutRecord exemplo
O código a seguir cria dez registros de dados, distribuídos entre duas chaves de partição, e os coloca em um fluxo denominado `myStreamName`.
```
for (int j = 0; j < 10; j++)
{
PutRecordRequest putRecordRequest = new PutRecordRequest();
putRecordRequest.setStreamName( myStreamName );
putRecordRequest.setData(ByteBuffer.wrap( String.format( "testData-%d", j ).getBytes() ));
putRecordRequest.setPartitionKey( String.format( "partitionKey-%d", j/5 ));
putRecordRequest.setSequenceNumberForOrdering( sequenceNumberOfPreviousRecord );
PutRecordResult putRecordResult = client.putRecord( putRecordRequest );
sequenceNumberOfPreviousRecord = putRecordResult.getSequenceNumber();
}
```
O código de exemplo anterior usa `setSequenceNumberForOrdering` para garantir estritamente o aumento da ordenação dentro de cada chave de partição. Para usar esse parâmetro de forma eficaz, defina o `SequenceNumberForOrdering` do registro atual (registro *n*) como o número de sequência do registro anterior (registro *n-1*). Para obter o número sequencial de um registro que foi adicionado ao fluxo, chame `getSequenceNumber` para o resultado de `putRecord`.
O parâmetro `SequenceNumberForOrdering` garante estritamente o aumento de números de sequência para a mesma chave de partição. `SequenceNumberForOrdering` não fornece a ordenação de registros em várias chaves de partição.
# Interaja com os dados usando o AWS Glue Schema Registry
Você pode integrar seus streams de dados do Kinesis com o AWS Glue Schema Registry. O registro de esquemas do AWS Glue permite detectar, controlar e evoluir esquemas centralmente, ao mesmo tempo que garante que os dados produzidos sejam validados continuamente por um esquema registrado. O esquema define a estrutura e o formato de um registro de dados. Um esquema é uma especificação versionada para publicação, consumo ou datastore confiáveis. O AWS Glue Schema Registry permite que você melhore a qualidade end-to-end dos dados e a governança de dados em seus aplicativos de streaming. Para obter mais informações, consulte [Registro de esquemas do AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html). Uma das formas de configurar essa integração é por meio do `PutRecords` `PutRecord` Kinesis APIs Data Streams disponível AWS no Java SDK.
Para obter instruções detalhadas sobre como configurar a integração do Kinesis Data Streams com o registro do esquema usando o e o PutRecord Kinesis Data Streams, consulte a seção “Interagindo com dados PutRecords usando o APIs Kinesis Data Streams” [em Caso de uso: Integração](https://docs.aws.amazon.com/glue/latest/dg/schema-registry-integrations.html#schema-registry-integrations-kds) do Amazon Kinesis Data APIs Streams com o Glue Schema Registry. AWS
# Gravar no Amazon Kinesis Data Streams usando o Kinesis Agent
O Kinesis Agent é uma aplicação de software Java independente que oferece uma maneira fácil de coletar e enviar dados ao Kinesis Data Streams. O agente monitora continuamente um conjunto de arquivos e envia novos dados ao fluxo. Ele manipula o rodízio de arquivos, os pontos de verificação e as novas tentativas após falhas. Os dados são entregues de maneira confiável, imediata e simples. Ele também emite CloudWatch métricas da Amazon para ajudar você a monitorar e solucionar melhor o processo de streaming.
Por padrão, os registros são analisados em cada arquivo com base no caractere de nova linha (`'\n'`). No entanto, o agente também pode ser configurado para analisar registros de várias linhas (consulte [Especificar as definições da configuração do agente](#agent-config-settings)).
É possível instalar o agente em ambientes de servidor baseados no Linux, como servidores web, servidores de log e servidores de banco de dados. Após instalar o agente, configure-o especificando os arquivos a serem monitorados e o fluxo dos dados. Depois que o agente é configurado, ele coleta dados dos arquivos de forma durável e os envia confiavelmente ao fluxo.
**Topics**
+ [Concluir os pré-requisitos do Kinesis Agent](#prereqs)
+ [Fazer download e instalar o agente](#download-install)
+ [Configuração e inicialização do agente](#config-start)
+ [Especificar as definições da configuração do agente](#agent-config-settings)
+ [Monitorar vários diretórios de arquivos e gravação em vários fluxos](#sim-writes)
+ [Uso do agente para pré-processar dados](#pre-processing)
+ [Usar comandos da CLI do agente](#cli-commands)
+ [Perguntas frequentes](#agent-faq)
## Concluir os pré-requisitos do Kinesis Agent
+ O sistema operacional deve ser a AMI do Amazon Linux versão 2015.09 ou posterior ou o Red Hat Enterprise Linux versão 7 ou posterior.
+ Se estiver usando o Amazon EC2 para executar o agente, inicie a instância do EC2.
+ Gerencie suas AWS credenciais usando um dos seguintes métodos:
+ Especifique um perfil do IAM ao executar a instância do EC2.
+ Especifique AWS as credenciais ao configurar o agente (consulte [awsAccessKeyID](#awsAccessKeyId) e [awsSecretAccesschave](#awsSecretAccessKey)).
+ Edite `/etc/sysconfig/aws-kinesis-agent` para especificar sua região e suas chaves de AWS acesso.
+ [Se sua instância do EC2 estiver em uma AWS conta diferente, crie uma função do IAM para fornecer acesso ao serviço Kinesis Data Streams e especifique essa função ao configurar o agente (consulte [AssumeroLearn](#assumeRoleARN) e Id). assumeRoleExternal](#assumeRoleExternalId) Use um dos métodos anteriores para especificar AWS as credenciais de um usuário na outra conta que tenha permissão para assumir essa função.
+ A função ou AWS as credenciais do IAM que você especificar devem ter permissão para realizar a operação do Kinesis Data [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)Streams para que o agente envie dados para seu stream. Se você ativar o CloudWatch monitoramento para o agente, a permissão para realizar a CloudWatch [PutMetricData](https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_PutMetricData.html)operação também será necessária. Para obter mais informações, consulte [Controlar o acesso aos recursos do Amazon Kinesis Data Streams usando o IAM](controlling-access.md)[Monitore a integridade do Kinesis Data Streams Agent com a Amazon CloudWatch](agent-health.md), e [Controle de CloudWatch acesso](https://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/UsingIAM.html).
## Fazer download e instalar o agente
Primeiro, conecte-se à instância. Para obter mais informações, consulte [Conectar-se à sua instância](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-connect-to-instance-linux.html) no *Guia do usuário do Amazon EC2*. Se tiver problemas para se conectar, consulte [Solução de problemas para conectar-se à sua instância](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/TroubleshootingInstancesConnecting.html) no *Guia do usuário do Amazon EC2*.
**Como configurar o agente usando o Amazon Linux AMI**
Use o comando a seguir para fazer download do agente e instalá-lo:
```
sudo yum install –y aws-kinesis-agent
```
**Como configurar o agente usando o Red Hat Enterprise Linux**
Use o comando a seguir para fazer download do agente e instalá-lo:
```
sudo yum install –y https://s3.amazonaws.com/streaming-data-agent/aws-kinesis-agent-latest.amzn2.noarch.rpm
```
**Para configurar o agente usando GitHub**
1. Baixe o agente em [amazon-kinesis-agentawlabs/](https://github.com/awslabs/amazon-kinesis-agent).
1. Instale o agente navegando até o diretório de download e executando o comando a seguir:
```
sudo ./setup --install
```
**Como configurar o agente em um contêiner do Docker**
O Kinesis Agent também pode ser executado em um contêiner por meio da base de contêineres [amazonlinux](https://docs.aws.amazon.com/AmazonECR/latest/userguide/amazon_linux_container_image.html). Use o Dockerfile a seguir e depois execute o `docker build`.
```
FROM amazonlinux
RUN yum install -y aws-kinesis-agent which findutils
COPY agent.json /etc/aws-kinesis/agent.json
CMD ["start-aws-kinesis-agent"]
```
## Configuração e inicialização do agente
**Como configurar e iniciar o agente**
1. Abra e edite o arquivo de configuração (como superusuário, se as permissões padrão de acesso a arquivos estiverem sendo usadas): `/etc/aws-kinesis/agent.json`
Nesse arquivo de configuração, especifique os arquivos ( `"filePattern"` ) nos quais o agente coleta dados e o nome do fluxo ( `"kinesisStream"` ) ao qual o agente envia dados. Observe que o nome do arquivo é um padrão, e o agente reconhece os rodízios de arquivos. Só é possível fazer o rodízio de arquivos ou criar novos arquivos uma vez por segundo, no máximo. O agente usa o carimbo de data e hora de criação de arquivo para determinar quais arquivos serão rastreados e colocados no final do fluxo; a criação de novos arquivos ou o rodízio de arquivos em uma frequência superior a uma vez por segundo não permite que o agente faça a distinção entre eles corretamente.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "yourkinesisstream"
}
]
}
```
1. Inicie o agente manualmente:
```
sudo service aws-kinesis-agent start
```
1. (Opcional) Configure o agente para ser iniciado durante o startup do sistema:
```
sudo chkconfig aws-kinesis-agent on
```
Agora o agente está sendo executado como um serviço do sistema em segundo plano. Ele monitora continuamente os arquivos especificados e envia dados ao fluxo especificado. A atividade do agent é registrada em `/var/log/aws-kinesis-agent/aws-kinesis-agent.log`.
## Especificar as definições da configuração do agente
O agente oferece suporte a duas configurações obrigatórias, `filePattern` e `kinesisStream`, além das configurações opcionais de recursos adicionais. É possível especificar configurações obrigatórias e opcionais em `/etc/aws-kinesis/agent.json`.
Sempre que o arquivo de configuração for alterado, o agente deverá ser interrompido e iniciado, usando os seguintes comandos:
```
sudo service aws-kinesis-agent stop
sudo service aws-kinesis-agent start
```
Se desejar, é possível usar o comando a seguir:
```
sudo service aws-kinesis-agent restart
```
Estas são as configurações gerais.
| Definição da configuração | Description |
| --- | --- |
| assumeRoleARN | O ARN da função a ser assumida pelo usuário. Para obter mais informações, consulte [Delegar acesso entre AWS contas usando funções do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/tutorial_cross-account-with-roles.html) no *Guia do usuário do IAM*. |
| assumeRoleExternalId | Um identificador opcional que determina quem pode assumir o perfil. Para obter mais informações, consulte [Como usar um ID externo](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user_externalid.html) no *Guia do usuário do IAM*. |
| awsAccessKeyId | AWS ID da chave de acesso que substitui as credenciais padrão. Essa configuração tem precedência sobre todos os outros provedores de credenciais. |
| awsSecretAccessKey | AWS chave secreta que substitui as credenciais padrão. Essa configuração tem precedência sobre todos os outros provedores de credenciais. |
| cloudwatch.emitMetrics | Permite que o agente emita métricas para, CloudWatch se definidas (verdadeiras). Padrão: verdadeiro |
| cloudwatch.endpoint | O endpoint regional para CloudWatch. Padrão: `monitoring.us-east-1.amazonaws.com` |
| kinesis.endpoint | O endpoint regional do Kinesis Data Streams. Padrão: `kinesis.us-east-1.amazonaws.com` |
Estas são as configurações de fluxo.
| Definição da configuração | Description |
| --- | --- |
| dataProcessingOptions | A lista das opções de processamento aplicadas a cada registro analisado antes que ele seja enviado ao fluxo. As opções de processamento são executadas na ordem especificada. Para obter mais informações, consulte [Uso do agente para pré-processar dados](#pre-processing). |
| kinesisStream | [Obrigatório] O nome do fluxo. |
| filePattern | [Obrigatório] O diretório e o padrão de arquivo que devem ser combinados para serem coletados pelo agente. Para todos os arquivos correspondentes a esse padrão, deve ser concedida uma permissão de leitura a `aws-kinesis-agent-user`. Para o diretório que contém os arquivos, devem ser concedidas permissões de leitura e execução a `aws-kinesis-agent-user`. |
| initialPosition | A posição em que o arquivo começou a ser analisado. Os valores válidos são `START_OF_FILE` e `END_OF_FILE`. Padrão: `END_OF_FILE` |
| maxBufferAgeMillis | O tempo máximo, em milissegundos, durante o qual o agente armazena os dados em buffer antes de enviá-los ao fluxo. Intervalo de valores: 1.000 a 900.000 (1 segundo a 15 minutos) Padrão: 60.000 (1 minuto) |
| maxBufferSizeBytes | O tamanho máximo, em bytes, durante o qual o agente armazena os dados em buffer antes de enviá-los ao fluxo. Intervalo de valores: 1 a 4.194.304 (4 MB) Padrão: 4.194.304 (4 MB) |
| maxBufferSizeRecords | O número máximo de registros para os quais o agente armazena os dados em buffer antes de enviá-los ao fluxo. Intervalo de valores: 1 a 500 Padrão: 500 |
| minTimeBetweenFilePollsMillis | O intervalo de tempo, em milissegundos, em que o agente consulta e analisa os arquivos monitorados em busca de novos dados. Intervalo de valores: 1 ou mais Padrão: 100 |
| multiLineStartPattern | O padrão de identificação do início de um registro. Um registro é composto por uma linha que corresponde ao padrão e pelas linhas subsequentes que não correspondem ao padrão. Os valores válidos são expressões regulares. Por padrão, cada nova linha nos arquivos de log é analisada como um único registro. |
| partitionKeyOption | O método para gerar a chave de partição. Os valores válidos são `RANDOM` (inteiro gerado aleatoriamente) e `DETERMINISTIC` (um valor de hash calculado a partir dos dados). Padrão: `RANDOM` |
| skipHeaderLines | O número de linhas em que o agente ignorará a análise no início dos arquivos monitorados. Intervalo de valores: 0 ou mais Padrão: 0 (zero) |
| truncatedRecordTerminator | A string que o agente usa para truncar um registro analisado que excede o limite de tamanho de registro do Kinesis Data Streams. (1,000 KB) Padrão: `'\n'` (nova linha) |
## Monitorar vários diretórios de arquivos e gravação em vários fluxos
Ao especificar vários fluxos de configurações, é possível configurar o agente para monitorar vários diretórios de arquivos e enviar dados a vários streams. No exemplo de configuração a seguir, o agente monitora dois diretórios de arquivos e envia dados para um fluxo do Kinesis e para um fluxo de entrega do Firehose, respectivamente. Observe que, como é possível especificar endpoints diferentes para o Kinesis Data Streams e o Firehose, os fluxos dos dois serviços não precisam estar na mesma região.
```
{
"cloudwatch.emitMetrics": true,
"kinesis.endpoint": "https://your/kinesis/endpoint",
"firehose.endpoint": "https://your/firehose/endpoint",
"flows": [
{
"filePattern": "/tmp/app1.log*",
"kinesisStream": "yourkinesisstream"
},
{
"filePattern": "/tmp/app2.log*",
"deliveryStream": "yourfirehosedeliverystream"
}
]
}
```
Para obter informações mais detalhadas sobre o uso do agente com o Firehose, consulte [Gravar no Amazon Kinesis Data Firehose com o Kinesis Agent](https://docs.aws.amazon.com/firehose/latest/dev/writing-with-agents.html).
## Uso do agente para pré-processar dados
O agente pode pré-processar os registros analisados a partir dos arquivos monitorados antes de enviá-los ao fluxo. É possível habilitar esse recurso adicionando a configuração `dataProcessingOptions` ao fluxo de arquivos. Um ou mais opções de processamento podem ser adicionadas e serão executadas na ordem especificada.
O agente oferece suporte às seguintes opções de processamento. Como o agente é de código aberto, é possível desenvolver e estender ainda mais suas opções de processamento. O download do agente pode ser feito em [Kinesis Agent](https://github.com/awslabs/amazon-kinesis-agent).Opções de processamento
`SINGLELINE`
Converte um registro de várias linhas em um registro de única linha removendo caracteres de nova linha, e espaços à esquerda e à direita.
```
{
"optionName": "SINGLELINE"
}
```
`CSVTOJSON`
Converte um registro com formato separado por delimitador em um registro com formato JSON.
```
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", ... ],
"delimiter": "yourdelimiter"
}
```
`customFieldNames`
[Obrigatório] Os nomes de campo usados como chaves em cada par de valores de chave JSON. Por exemplo, ao especificar `["f1", "f2"]`, o registro "v1, v2" será convertido em `{"f1":"v1","f2":"v2"}`.
`delimiter`
A string usada como delimitador no registro. O padrão é uma vírgula (,).
`LOGTOJSON`
Converte um registro com formato de log em um registro com formato JSON. Os formatos de log com suporte são **Apache Common Log**, **Apache Combined Log**, **Apache Error Log** e **RFC3164 Syslog**.
```
{
"optionName": "LOGTOJSON",
"logFormat": "logformat",
"matchPattern": "yourregexpattern",
"customFieldNames": [ "field1", "field2", … ]
}
```
`logFormat`
[Obrigatório] O formato da entrada de log. Os valores possíveis são:
+ `COMMONAPACHELOG`: o formato do Apache Common Log. Cada entrada de log tem o seguinte padrão: "`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes}`".
+ `COMBINEDAPACHELOG`: o formato do Apache Combined Log. Cada entrada de log tem o seguinte padrão: "`%{host} %{ident} %{authuser} [%{datetime}] \"%{request}\" %{response} %{bytes} %{referrer} %{agent}`".
+ `APACHEERRORLOG`: o formato do Apache Error Log. Cada entrada de log tem o seguinte padrão: "`[%{timestamp}] [%{module}:%{severity}] [pid %{processid}:tid %{threadid}] [client: %{client}] %{message}`".
+ `SYSLOG`— O formato RFC3164 Syslog. Cada entrada de log tem o seguinte padrão: "`%{timestamp} %{hostname} %{program}[%{processid}]: %{message}`".
`matchPattern`
O padrão da expressão regular usada para extrair valores de entradas de log. Essa configuração é usada se a entrada de log não estiver em um dos formatos de log predefinidos. Se essa configuração for usada, também é necessário especificar `customFieldNames`.
`customFieldNames`
Os nomes de campo personalizados usados como chaves em cada par de valores de chave JSON. É possível usar essa configuração para definir nomes de campo para valores extraídos de `matchPattern` ou substituir os nomes de campo padrão de formatos de log predefinidos.
**Example : Configuração LOGTOJSON**
Este é um exemplo de uma configuração `LOGTOJSON` para uma entrada Apache Common Log convertida em formato JSON:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
```
Antes da conversão:
```
64.242.88.10 - - [07/Mar/2004:16:10:02 -0800] "GET /mailman/listinfo/hsdivision HTTP/1.1" 200 6291
```
Depois da conversão:
```
{"host":"64.242.88.10","ident":null,"authuser":null,"datetime":"07/Mar/2004:16:10:02 -0800","request":"GET /mailman/listinfo/hsdivision HTTP/1.1","response":"200","bytes":"6291"}
```
**Example : Configuração LOGTOJSON com campos personalizados**
Este é outro exemplo de configuração `LOGTOJSON`:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"customFieldNames": ["f1", "f2", "f3", "f4", "f5", "f6", "f7"]
}
```
Com essa configuração, a mesma entrada Apache Common Log do exemplo anterior é convertida em formato JSON, da seguinte forma:
```
{"f1":"64.242.88.10","f2":null,"f3":null,"f4":"07/Mar/2004:16:10:02 -0800","f5":"GET /mailman/listinfo/hsdivision HTTP/1.1","f6":"200","f7":"6291"}
```
**Example : Conversão da entrada Apache Common Log**
A configuração de fluxo a seguir converte uma entrada Apache Common Log em um registro de linha única no formato JSON:
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"dataProcessingOptions": [
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG"
}
]
}
]
}
```
**Example : Conversão de registros de várias linhas**
A configuração de fluxo a seguir analisa registros de várias linha cuja primeira linha começa com "`[SEQUENCE=`". Cada registro é convertido primeiro em um registro de única linha. Em seguida, os valores são extraídos do registro com base em um delimitador por tabulações. Os valores extraídos são mapeados para os valores `customFieldNames` especificados, a fim de formar um registro de linha única no formato JSON.
```
{
"flows": [
{
"filePattern": "/tmp/app.log*",
"kinesisStream": "my-stream",
"multiLineStartPattern": "\\[SEQUENCE=",
"dataProcessingOptions": [
{
"optionName": "SINGLELINE"
},
{
"optionName": "CSVTOJSON",
"customFieldNames": [ "field1", "field2", "field3" ],
"delimiter": "\\t"
}
]
}
]
}
```
**Example : Configuração LOGTOJSON com padrão de correspondência**
este é um exemplo de configuração `LOGTOJSON` referente a uma entrada Apache Common Log convertida em formato JSON, com o último campo (bytes) omitido:
```
{
"optionName": "LOGTOJSON",
"logFormat": "COMMONAPACHELOG",
"matchPattern": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(.+?)\" (\\d{3})",
"customFieldNames": ["host", "ident", "authuser", "datetime", "request", "response"]
}
```
Antes da conversão:
```
123.45.67.89 - - [27/Oct/2000:09:27:09 -0400] "GET /java/javaResources.html HTTP/1.0" 200
```
Depois da conversão:
```
{"host":"123.45.67.89","ident":null,"authuser":null,"datetime":"27/Oct/2000:09:27:09 -0400","request":"GET /java/javaResources.html HTTP/1.0","response":"200"}
```
## Usar comandos da CLI do agente
Inicie automaticamente o agente durante o startup do sistema:
```
sudo chkconfig aws-kinesis-agent on
```
Verifique o status do agente:
```
sudo service aws-kinesis-agent status
```
Interrompa o agente:
```
sudo service aws-kinesis-agent stop
```
Leia o arquivo de log do agente a partir deste local:
```
/var/log/aws-kinesis-agent/aws-kinesis-agent.log
```
Desinstale o agente:
```
sudo yum remove aws-kinesis-agent
```
## Perguntas frequentes
### Existe um Kinesis Agent para Windows?
O [Kinesis Agent para Windows](https://docs.aws.amazon.com/kinesis-agent-windows/latest/userguide/what-is-kinesis-agent-windows.html) é um software diferente das plataformas do Kinesis Agent para Linux.
### Por que a desaceleração do Kinesis Agent está aumentando? and/or `RecordSendErrors`
Isso geralmente ocorre devido ao controle de utilização do Kinesis. Verifique a métrica `WriteProvisionedThroughputExceeded` do Kinesis Data Streams ou a métrica `ThrottledRecords` dos fluxos de entrega do Firehose. Qualquer aumento de 0 nessas métricas indica que os limites do fluxo precisam ser aumentados. Para obter mais informações, consulte [Kinesis Data Stream limits](https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html) e [Amazon Firehose Delivery Streams](https://docs.aws.amazon.com/firehose/latest/dev/limits.html).
Depois de descartar o controle de utilização como causa, verifique se o Kinesis Agent está configurado para seguir um número grande de arquivos pequenos. Há um atraso quando o Kinesis Agent exibe os dados do final de um arquivo novo, portanto, o Kinesis Agent deveria estar exibindo os dados do final de um pequeno número de arquivos maiores. Tente consolidar os arquivos de log em arquivos maiores.
### Por que estou recebendo exceções `java.lang.OutOfMemoryError`?
O Kinesis Agent não tem memória suficiente para lidar com a workload atual. Tente aumentar `JAVA_START_HEAP` e `JAVA_MAX_HEAP` no `/usr/bin/start-aws-kinesis-agent` e reiniciar o agente.
### Por que estou recebendo exceções `IllegalStateException : connection pool shut down`?
O Kinesis Agent não tem conexões suficientes para lidar com a workload atual. Tente aumentar `maxConnections` e `maxSendingThreads` nas configurações gerais do agente em `/etc/aws-kinesis/agent.json`. O valor padrão para esses campos é 12 vezes o número de processadores de runtime disponíveis. Consulte [AgentConfiguration.java](https://github.com/awslabs/amazon-kinesis-agent/blob/master/src/com/amazon/kinesis/streaming/agent/config/AgentConfiguration.java) para saber mais sobre as configurações avançadas do agente.
### Como posso depurar outro problema com o Kinesis Agent?
Os logs do nível `DEBUG` podem ser habilitados em `/etc/aws-kinesis/log4j.xml`.
### Como devo configurar o Kinesis Agent?
Quanto menor o `maxBufferSizeBytes`, mais frequentemente o Kinesis Agent enviará dados. Isso pode ser bom, pois diminui o tempo de entrega dos registros, mas também aumenta as solicitações por segundo feitas ao Kinesis.
### Por que o Kinesis Agent está enviando registros duplicados?
Isso ocorre devido a uma configuração incorreta da exibição dos dados do final dos arquivos. Certifique-se de que cada `fileFlow’s filePattern` corresponda a apenas um arquivo. Isso também pode ocorrer se o modo `logrotate` que está sendo usado estiver no modo `copytruncate`. Tente mudar o modo para o modo padrão ou criar para evitar duplicações. Para obter mais informações sobre como lidar com registros duplicados, consulte [Handling Duplicate Records](https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-duplicates.html).
# Grave no Kinesis Data Streams usando outros serviços AWS
Os AWS serviços a seguir podem se integrar diretamente ao Amazon Kinesis Data Streams para gravar dados nos streams de dados do Kinesis. Leia as informações sobre cada serviço que seja de seu interesse e consulte as referências fornecidas.
**Topics**
+ [Grave no Kinesis Data Streams usando AWS Amplify](using-other-services-amplify.md)
+ [Gravar no Kinesis Data Streams usando o Amazon Aurora](using-other-services-aurora.md)
+ [Escreva para o Kinesis Data Streams usando a Amazon CloudFront](using-other-services-CloudFront.md)
+ [Grave no Kinesis Data Streams CloudWatch usando o Amazon Logs](using-other-services-cw-logs.md)
+ [Gravar no Kinesis Data Streams usando o Amazon Connect](using-other-services-connect.md)
+ [Grave no Kinesis Data Streams usando AWS Database Migration Service](using-other-services-migration.md)
+ [Gravar no Kinesis Data Streams usando o Amazon DynamoDB](using-other-services-ddb.md)
+ [Escreva para o Kinesis Data Streams usando a Amazon EventBridge](using-other-services-eventbridges.md)
+ [Grave no Kinesis Data Streams usando AWS IoT Core](using-other-services-iot-core.md)
+ [Gravar no Kinesis Data Streams usando o Amazon Relational Database Service](using-other-services-rds.md)
+ [Gravar no Kinesis Data Streams usando o Amazon Pinpoint](using-other-services-pinpoint.md)
+ [Gravar no Kinesis Data Streams usando o Amazon Quantum Ledger Database (Amazon QLDB)](using-other-services-quantum-ledger.md)
# Grave no Kinesis Data Streams usando AWS Amplify
Você pode usar o Amazon Kinesis Data Streams para transmitir dados de aplicações móveis criadas com o AWS Amplify para processamento em tempo real. Isso permite a criação de painéis em tempo real, capture exceções, gere alertas, estimule recomendações e tome outras decisões comerciais ou operacionais oportunas. Você também pode enviar dados para outros serviços, como o Amazon Simple Storage Service, o Amazon DynamoDB e o Amazon Redshift.
Para obter mais informações, consulte [Using Amazon Kinesis](https://docs.amplify.aws/react/build-a-backend/more-features/analytics/streaming-data/) no *AWS Amplify Developer Center*.
# Gravar no Kinesis Data Streams usando o Amazon Aurora
É possível usar o Amazon Kinesis Data Streams para monitorar atividades em clusters de banco de dados do Amazon Aurora. Usando o Database Activity Streams, seu cluster de banco de dados do Aurora envia atividades para um fluxo de dados do Amazon Kinesis em tempo real. Em seguida, pode-se criar aplicações de gerenciamento de conformidade para consumir essas atividades, auditá-las e gerar alertas. Também é possível usar o Amazon Firehose para armazenar dados.
Para obter mais informações, consulte [Database Activity Streams](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/DBActivityStreams.html) no *Guia do desenvolvedor do Amazon Aurora*.
# Escreva para o Kinesis Data Streams usando a Amazon CloudFront
Você pode usar o Amazon Kinesis Data CloudFront Streams com registros em tempo real e obter informações sobre solicitações feitas para uma distribuição em tempo real. Em seguida, você pode criar seu próprio [consumidor de stream de dados do Kinesis](https://docs.aws.amazon.com/streams/latest/dev/building-consumers.html) ou usar o Amazon Data Firehose para enviar os dados de log para o Amazon S3, Amazon Redshift, Amazon Service ou um serviço de processamento de log de OpenSearch terceiros.
Para obter mais informações, consulte [Registros em tempo real](https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/real-time-logs.html) no *Amazon CloudFront Developer Guide*.
# Grave no Kinesis Data Streams CloudWatch usando o Amazon Logs
Você pode usar CloudWatch assinaturas para ter acesso a um feed em tempo real de eventos de log do Amazon CloudWatch Logs e entregá-lo a um stream de dados do Kinesis para processamento, análise e carregamento em outros sistemas.
Para obter mais informações, consulte [Processamento em tempo real de dados de log com assinaturas no Guia](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Subscriptions.html) *do usuário do Amazon CloudWatch Logs*.
# Gravar no Kinesis Data Streams usando o Amazon Connect
É possível usar o Kinesis Data Streams para exportar registros de contatos e eventos de agentes em tempo real da sua instância do Amazon Connect. Você também pode habilitar o fluxo de dados do Amazon Connect Customer Profiles para receber automaticamente atualizações em um fluxo de dados do Kinesis sobre a criação de novos perfis ou alterações nos perfis existentes.
Em seguida, você pode criar aplicações de consumo para processar e analisar os dados em tempo real. Por exemplo, usando registros de contato e dados do perfil do cliente, você pode manter os dados do sistema de origem, como CRMs ferramentas de automação de marketing, up-to-date com as informações mais recentes. Usando os dados de eventos do agente, é possível criar painéis que exibem informações e eventos, além de acionar notificações personalizadas de atividades específicas do agente.
Para obter mais informações, consulte [Fluxo de dados para sua instância](https://docs.aws.amazon.com/connect/latest/adminguide/data-streaming.html), [Configurar uma exportação em tempo real](https://docs.aws.amazon.com/connect/latest/adminguide/set-up-real-time-export.html) e [Fluxos de eventos de agente](https://docs.aws.amazon.com/connect/latest/adminguide/agent-event-streams.html) no *Guia do administrador do Amazon Connect*.
# Grave no Kinesis Data Streams usando AWS Database Migration Service
Você pode usar AWS Database Migration Service para migrar dados para um stream de dados do Kinesis. Em seguida, é possível criar aplicações de consumo para processar os registros de dados em tempo real. Você também pode enviar facilmente dados downstream para outros serviços, como o Amazon Simple Storage Service, o Amazon DynamoDB e o Amazon Redshift.
Para obter mais informações, consulte [Using Kinesis Data Streams](https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html) no *Guia do usuário do AWS Database Migration Service *.
# Gravar no Kinesis Data Streams usando o Amazon DynamoDB
É possível usar o Amazon Kinesis Data Streams para capturar alterações no Amazon DynamoDB. O Kinesis Data Streams capta alterações no nível de item em qualquer DynamoDB e as replica em um fluxo de dados do Kinesis. Suas aplicações de consumo podem acessar esse fluxo para visualizar as alterações nos itens em tempo real e entregá-las downstream ou realizar ações com base no seu conteúdo.
Para obter mais informações, consulte [Como o Kinesis Data Streams funciona com o DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/kds.html) no *Guia do desenvolvedor do Amazon DynamoDB*.
# Escreva para o Kinesis Data Streams usando a Amazon EventBridge
Usando o Kinesis Data Streams, você AWS pode enviar EventBridge eventos de chamada de [API](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-events.html) para um stream, criar aplicativos de consumo e processar grandes quantidades de dados. Você também pode usar o Kinesis Data Streams como EventBridge destino no Pipes e entregar registros de um stream de uma das fontes disponíveis após filtragem e enriquecimento opcionais.
Para obter mais informações, consulte [Enviar eventos para um stream do Amazon Kinesis](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-relay-events-kinesis-stream.html) e [EventBridge Pipes no Guia EventBridge ](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html) *do usuário da Amazon*.
# Grave no Kinesis Data Streams usando AWS IoT Core
Você pode gravar dados em tempo real a partir de mensagens MQTT no AWS IoT Core usando ações de regras de AWS IoT. Em seguida, pode-se criar aplicações para processar os dados, analisar seu conteúdo, gerar alertas e entregar os dados a aplicações de analytics ou outros serviços da AWS .
Para obter mais informações, consulte [Kinesis Data Streams](https://docs.aws.amazon.com/iot/latest/developerguide/kinesis-rule-action.html) no *Guia do desenvolvedor do AWS IoT Core*.
# Gravar no Kinesis Data Streams usando o Amazon Relational Database Service
É possível usar o Amazon Kinesis Data Streams para monitorar atividades em instâncias do Amazon RDS. Usando o Database Activity Streams, o Amazon RDS envia atividades para um fluxo de dados do Kinesis em tempo real. Em seguida, pode-se criar aplicações de gerenciamento de conformidade para consumir essas atividades, auditá-las e gerar alertas. Também é possível usar o Amazon Data Firehose para armazenar dados.
Para obter mais informações, consulte [Database Activity Streams](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/DBActivityStreams.html) no *Guia do desenvolvedor do Amazon RDS*.
# Gravar no Kinesis Data Streams usando o Amazon Pinpoint
É possível configurar o Amazon Pinpoint para enviar dados de eventos ao Amazon Kinesis Data Streams. O Amazon Pinpoint pode enviar dados de eventos para campanhas, viagens e mensagens de e-mail e SMS transacionais. Em seguida, é possível ingerir os dados em aplicações de analytics ou criar as próprias aplicações de consumo para realizar ações com base no conteúdo dos eventos.
Para obter mais informações, consulte [Streaming Events](https://docs.aws.amazon.com/pinpoint/latest/developerguide/event-streams.html) no *Guia do desenvolvedor do Amazon Pinpoint*.
# Gravar no Kinesis Data Streams usando o Amazon Quantum Ledger Database (Amazon QLDB)
Você pode criar um fluxo no Amazon QLDB para capturar todas as revisões de documentos confirmadas no diário e entregar esses dados ao Amazon Kinesis Data Streams em tempo real. Assim, o QLDB mantém um fluxo contínuo de dados do diário do livro-razão para um recurso de fluxo de dados do Kinesis. Em seguida, é possível usar a plataforma de fluxo Kinesis ou a Kinesis Client Library para consumir o fluxo, processar os registros de dados e analisar o conteúdo dos dados. Um fluxo do QLDB grava dados no Kinesis Data Streams em três tipos de registros `control`, `block summary` e `revision details`.
Para obter mais informações, consulte [Streams](https://docs.aws.amazon.com/qldb/latest/developerguide/streams.html) no *Guia do desenvolvedor do Amazon QLDB*.
# Gravar no Kinesis Data Streams usando integrações de terceiros
Você pode gravar dados no Kinesis Data Streams usando uma das opções de terceiros a seguir que se integram ao Kinesis Data Streams: Selecione uma opção para saber mais sobre ela e encontre recursos e links para a documentação relevante.
**Topics**
+ [Apache Flink](using-other-services-flink.md)
+ [Fluentd](using-other-services-Fluentd.md)
+ [Debezium](using-other-services-Debezium.md)
+ [Oráculo GoldenGate](using-other-services-Oracle-GoldenGate.md)
+ [Kafka Connect](using-other-services-kafka-connect.md)
+ [Adobe Experience](using-other-services-adobe.md)
+ [Striim](using-other-services-Striim.md)
# Apache Flink
O Apache Flink é um framework de código aberto distribuído para computações com estado em fluxos de dados delimitados e não delimitados. Para obter mais informações sobre como gravar no Kinesis Data Streams do Apache Flink, consulte [Amazon Kinesis Data Streams Connector](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/kinesis/).
# Fluentd
O Fluentd é um coletor de dados de código aberto para uma camada de registro unificada. Para obter mais informações sobre como gravar no Kinesis Data Streams do Fluentd, consulte [Stream processing with Kinesis](https://docs.fluentd.org/how-to-guides/kinesis-stream).
# Debezium
O Debezium é uma plataforma distribuída de código aberto para captura de mudanças nos dados. Para obter mais informações sobre como gravar no Kinesis Data Streams do Debezium, consulte [Streaming MySQL Data Changes to Amazon Kinesis](https://debezium.io/blog/2018/08/30/streaming-mysql-data-changes-into-kinesis/).
# Oráculo GoldenGate
GoldenGate O Oracle é um produto de software que permite replicar, filtrar e transformar dados de um banco de dados para outro. Para obter mais informações sobre como gravar no Kinesis Data Streams GoldenGate da Oracle[, consulte Replicação de dados para o Kinesis](https://blogs.oracle.com/dataintegration/post/data-replication-to-aws-kinesis-data-stream-using-oracle-goldengate) Data Stream usando Oracle. GoldenGate
# Kafka Connect
O Kafka Connect é uma ferramenta para transmitir dados de forma escalável e confiável entre o Apache Kafka e outros sistemas. Para obter mais informações sobre como gravar no Kinesis Data Streams do Apache Kafka, consulte [Kinesis kafka connector](https://github.com/awslabs/kinesis-kafka-connector).
# Adobe Experience
A Adobe Experience Platform permite que as organizações centralizem e padronizem dados dos clientes em qualquer sistema. Em seguida, a plataforma aplica ciência de dados e machine learning para melhorar significativamente o design e a entrega de experiências ricas e personalizadas. Para obter mais informações sobre como gravar dados no Kinesis Data Streams da Adobe Experience Platform, aprenda como criar uma [conexão com o Amazon Kinesis](https://experienceleague.adobe.com/docs/experience-platform/destinations/catalog/cloud-storage/amazon-kinesis.html?lang=en).
# Striim
O Striim é uma plataforma completa em memória para coletar, filtrar, transformar, enriquecer, agregar, analisar e fornecer dados em tempo real. end-to-end Para obter mais informações sobre como gravar dados no Kinesis Data Streams do Striim, consulte [Kinesis Writer](https://www.striim.com/docs/en/kinesis-writer.html).
# Solução de problemas de produtores do Amazon Kinesis Data Streams
**Topics**
+ [Meu aplicativo produtor está gravando a uma taxa menor que a esperada](#producer-writing-at-slower-rate)
+ [Eu recebo um erro de permissão de chave mestra do KMS não autorizada](#unauthorized-kms-producer)
+ [Solucionar outros problemas comuns para produtores](#misc-troubleshooting-producer)
## Meu aplicativo produtor está gravando a uma taxa menor que a esperada
**Topics**
+ [Limites de serviço excedidos](#service-limits-exceeded)
+ [Quero otimizar meu produtor](#producer-optimization)
+ [Uso indevido das operações `flushSync()`](#misuse-tag)
### Limites de serviço excedidos
Para descobrir se os limites de serviço estão sendo excedidos, verifique se o produtor está lançando exceções de throughput a partir do serviço e valide quais operações da API estão sendo aceleradas. Lembre-se de que há limites diferentes de acordo com a chamada, consulte [Cotas e limites](service-sizes-and-limits.md). Por exemplo, além dos limites de nível de fragmento para gravações e leituras que são mais comumente conhecidas, há limites de nível de fluxo a seguir:
+ [CreateStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_CreateStream.html)
+ [DeleteStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DeleteStream.html)
+ [ListStreams](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListStreams.html)
+ [GetShardIterator](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html)
+ [MergeShards](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_MergeShards.html)
+ [DescribeStream](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStream.html)
+ [DescribeStreamSummary](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStreamSummary.html)
As operações `CreateStream`, `DeleteStream`, `ListStreams`, `GetShardIterator` e `MergeShards` são limitadas a 5 chamadas por segundo. A operação `DescribeStream` é limitada a 10 chamadas por segundo. A operação `DescribeStreamSummary` é limitada a 20 chamadas por segundo.
Se essas chamadas não forem o problema, selecione uma chave de partição que permita distribuir operações *put* uniformemente em todos os fragmentos e não tenha uma determinada chave de partição que esteja colidindo com os limites de serviço quando as restantes não estão. Isso requer a medição do throughput de pico e que seja levado em conta o número de fragmentos no seu fluxo. Para obter mais informações sobre o gerenciamento de streams, consulte [Criar e gerenciar fluxos de dados do Kinesis](working-with-streams.md).
**dica**
Lembre-se de arredondar para o kilobyte mais próximo para cálculos de limitação de taxa de transferência ao usar a operação de um único registro [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html), enquanto a operação de vários registros [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html) é arredondada na soma cumulativa dos registros em cada chamada. Por exemplo, uma solicitação `PutRecords` com 600 registros com tamanho de 1,1 KB não serão aceleradas.
### Quero otimizar meu produtor
Antes de começar a otimizar o produtor, conclua as seguintes tarefas importantes. Primeiro, identifique seu throughput de pico desejado em termos de tamanho do registro e registros por segundo. Em seguida, descarte a capacidade de fluxo conforme o fator de limitação ([Limites de serviço excedidos](#service-limits-exceeded)). Se a capacidade de fluxo foi excluída, use as seguintes dicas de solução de problemas e diretrizes de otimização para os dois tipos comuns de aplicações de produção.
**Produtor grande**
Um grande produtor normalmente é executado em um servidor on-premises ou em uma instância do Amazon EC2. Os clientes que precisam de um throughput mais alto de um grande produtor normalmente se preocupam com a latência por registro. As estratégias para lidar com a latência incluem o seguinte: Se o cliente puder armazenar registros em microlote/buffer, use a [Amazon Kinesis Producer Library](https://docs.aws.amazon.com/kinesis/latest/dev/developing-producers-with-kpl.html) (que tem lógica de agregação avançada), a operação de vários registros ou agregar registros em um arquivo maior antes de usar a operação [PutRecords](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html)de registro único. [PutRecord](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html) Se não for possível criar microlotes ou armazenar registros em buffer, use vários threads para gravar no serviço Kinesis Data Streams ao mesmo tempo. Os AWS SDK para Java e outros SDKs incluem clientes assíncronos que podem fazer isso com muito pouco código.
**Produtor pequeno**
Um pequeno produtor geralmente é um aplicativo móvel, dispositivo IoT ou cliente web. Se for um aplicativo móvel, recomendamos usar a `PutRecords` operação ou o Kinesis Recorder no celular. AWS SDKs Para obter mais informações, consulte AWS Mobile SDK para Android Guia de introdução e AWS Mobile SDK for iOS Guia de introdução. Aplicativos móveis devem lidar com conexões intermitentes inerentemente e precisam de algum tipo de alocação em lote, como `PutRecords`. Se não for possível alocar em lote por algum motivo, consulte as informações sobre Grande produtor acima. Se o seu produtor é um navegador, a quantidade de dados que está sendo gerada geralmente é muito pequena. No entanto, as operações *put* estão sendo colocadas no caminho crítico do aplicativo, o que não é recomendável.
### Uso indevido das operações `flushSync()`
O uso incorreto do `flushSync()` pode afetar significativamente o desempenho de gravação. A operação do `flushSync()` foi projetada para cenários de desligamento, a fim de garantir que todos os registros armazenados em buffer sejam enviados antes que a aplicação KPL seja encerrada. Se você implementou essa operação após cada operação de gravação, ela pode adicionar uma latência extra substancial de cerca de 500 ms por gravação. Assegure que `flushSync()` tenha sido implementado somente para o desligamento da aplicação para evitar atrasos extras desnecessários no desempenho de gravação.
## Eu recebo um erro de permissão de chave mestra do KMS não autorizada
Esse erro ocorre quando um aplicativo produtor grava em um fluxo criptografado sem permissões na chave mestra do KMS. Para atribuir permissões a uma aplicação para que acesse uma chave do KMS, consulte [Using Key Policies in AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/key-policies.html) e [Using IAM Policies with AWS KMS](https://docs.aws.amazon.com/kms/latest/developerguide/iam-policies.html).
## Solucionar outros problemas comuns para produtores
+ [Por que meu fluxo de dados Kinesis retorna um Erro de Servidor Interno 500?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-500-error/)
+ [How do I troubleshoot timeout errors when writing from Flink to Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-flink-timeout/)
+ [How do I troubleshoot throttling errors in Kinesis Data Streams?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling-errors/)
+ [Por que ocorre controle de utilização em meu fluxo de dados Kinesis?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-throttling/)
+ [Como posso colocar registros de dados em um fluxo de dados usando a KPL?](https://aws.amazon.com/premiumsupport/knowledge-center/kinesis-data-stream-kpl/)
# Otimizar os produtores do Kinesis Data Streams
Você pode otimizar ainda mais seus produtores do Amazon Kinesis Data Streams dependendo do comportamento específico observado. Leia os seguintes tópicos para identificar soluções.
**Topics**
+ [Personalize o comportamento de limites de taxas e novas tentativas da KPL](kinesis-producer-adv-retries-rate-limiting.md)
+ [Aplique as práticas recomendadas à agregação da KPL](kinesis-producer-adv-aggregation.md)
# Personalize o comportamento de limites de taxas e novas tentativas da KPL
Ao adicionar registros de usuário do Amazon Kinesis Producer Library (KPL) usando a operação `addUserRecord()` da KPL, o registro recebe um carimbo de data e hora e é adicionado a um buffer com um prazo definido pelo parâmetro de configuração `RecordMaxBufferedTime`. Essa stamp/deadline combinação de tempo define a prioridade do buffer. Os registros são descarregados do buffer com base nos seguintes critérios:
+ Prioridade do buffer
+ Configuração de agregação
+ Configuração de coleta
Os parâmetros de configuração de coleta e agregação que afetam o comportamento do buffer são os seguintes:
+ `AggregationMaxCount`
+ `AggregationMaxSize`
+ `CollectionMaxCount`
+ `CollectionMaxSize`
Em seguida, os registros descarregados são enviados para o fluxo de dados do Kinesis como registros do Amazon Kinesis Data Streams usando uma chamada para a operação `PutRecords` de API do Kinesis Data Streams. A operação `PutRecords` envia solicitações ao fluxo que ocasionalmente apresenta falhas parciais ou totais. Os registros com falha são automaticamente adicionados de volta ao buffer da KPL. O novo prazo é definido com base no mínimo destes dois valores:
+ Metade da configuração de `RecordMaxBufferedTime` atual
+ O time-to-live valor do registro
Essa estratégia permite que registros repetidos de usuários do KPL sejam incluídos nas chamadas subsequentes da API do Kinesis Data Streams, para melhorar a taxa de transferência e reduzir a complexidade, ao mesmo tempo em que reforça o valor do registro do Kinesis Data Streams. time-to-live Não há um algoritmo de espera, tornando essa uma estratégia de tentativa relativamente agressiva. O spam devido ao excesso de tentativas é evitado pela limitação de taxas, discutida na próxima seção.
## Limitação de intervalo
A KPL inclui um recurso de limitação de taxas, que limita o throughput por fragmento enviado de um único produtor. A limitação de taxas é implementada usando um algoritmo de bucket de token com buckets separados para bytes e registros do Kinesis Data Streams. Cada gravação bem-sucedida em um fluxo de dados do Kinesis adiciona um token (ou vários tokens) a cada bucket, até um determinado limite. Esse limite é configurável, mas por padrão é definido como 50% maior que o limite de fragmento real, para permitir a saturação de fragmentos a partir de um único produtor.
É possível reduzir esse limite para diminuir o spam devido ao excesso de tentativas. No entanto, a melhor prática é que cada produtor tente novamente para um throughput máximo de maneira agressiva e lide com qualquer limitação resultante determinada como excessiva por meio da expansão da capacidade do fluxo e da implementação de uma estratégia de chave de partição apropriada.
# Aplique as práticas recomendadas à agregação da KPL
Embora o esquema do número de sequência dos registros resultantes do Amazon Kinesis Data Streams permaneça o mesmo, a agregação faz com que a indexação dos registros de usuários do Amazon Kinesis Producer Library (KPL) contidos em um registro agregado do Kinesis Data Streams comece de 0 (zero). No entanto, desde que não haja dependência dos números de sequência para identificar exclusivamente os registros de usuários da KPL, o código poderá ignorar isso, já que a agregação (de registros de usuários da KPL em um registro do Kinesis Data Streams) e a desagregação subsequente (de um registro do Kinesis Data Streams em registros de usuários da KPL) cuidam disso automaticamente para você. Isso se aplica se seu consumidor estiver usando o KCL ou o AWS SDK. Para usar essa funcionalidade de agregação, você precisará inserir a parte Java do KPL em sua compilação se seu consumidor for escrito usando a API fornecida no AWS SDK.
Se quiser usar números de sequência como identificadores exclusivos dos registros de usuários da KPL, recomendamos que use as operações `public int hashCode()` e `public boolean equals(Object obj)`, que respeitam contratos, fornecidas em `Record` e `UserRecord` para habilitar a comparação desses registros. Além disso, para examinar o número subsequente do registro de usuários da KPL, é possível convertê-lo em uma instância de `UserRecord` e recuperar o número de subsequência.
Para obter mais informações, consulte [Implementar a desagregação de consumidores](kinesis-kpl-consumer-deaggregation.md).