Crie e execute um serviço gerenciado para o aplicativo Apache Flink para Python

Como criar os fluxos de dados (AWS CLI)

1. Para criar o primeiro stream (ExampleInputStream), use o seguinte comando do Amazon Kinesis create-stream AWS CLI .

   $ aws kinesis create-stream \
   --stream-name ExampleInputStream \
   --shard-count 1 \
   --region us-east-1 

2. Para criar o segundo fluxo que o aplicativo usa para gravar a saída, execute o mesmo comando, alterando o nome da transmissão para ExampleOutputStream.

   $ aws kinesis create-stream \
   --stream-name ExampleOutputStream \
   --shard-count 1 \
   --region us-east-1 

1. Duplique o repositório remoto usando o seguinte comando:

   git clone https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples.git

2. Navegue até o diretório ./python/GettingStarted.

Para executar seu aplicativo a partir da linha de comando

1. Certifique-se de que o ambiente autônomo do Python, como o Conda VirtualEnv ou onde você instalou a biblioteca Python Flink, esteja ativo no momento.

2. Certifique-se de correr pelo mvn package menos uma vez.

3. Defina a IS_LOCAL = true variável de ambiente:

   $ export IS_LOCAL=true

4. Execute o aplicativo como um script Python normal.

   $python main.py

Para executar o aplicativo de dentro do IDE

1. Configure seu IDE para executar o main.py script com a seguinte configuração:

   1. Use o ambiente Python independente, como o Conda VirtualEnv ou onde você instalou a biblioteca. PyFlink

   2. Use as AWS credenciais para acessar os streams de dados de entrada e saída do Kinesis.

   3. Defina IS_LOCAL = true.

2. O processo exato para definir a configuração de execução depende de você IDE e varia.

3. Depois de configurar o seuIDE, execute o script Python e use as ferramentas fornecidas por você IDE enquanto o aplicativo estiver em execução.

Para carregar o JAR arquivo de código do aplicativo

1. Abra o console do Amazon S3 em. https://console.aws.amazon.com/s3/

2. Escolha o bucket que você criou anteriormente para o código do aplicativo.

3. Escolha Carregar.

4. Escolha Adicionar arquivos.

5. Navegue até o arquivo.zip gerado na etapa anterior:target/managed-flink-pyflink-getting-started-1.0.0.zip.

6. Escolha Carregar sem alterar nenhuma outra configuração.

1. Abra o console do Managed Service for Apache Flink em /flink https://console.aws.amazon.com

2. Verifique se a região correta está selecionada: Leste dos EUA (Norte da Virgínia) us-east-1.

3. Abra o menu do lado direito e escolha aplicativos Apache Flink e, em seguida, Criar aplicativo de streaming. Como alternativa, escolha Criar aplicativo de streaming na seção Começar da página inicial.

4. Na página Criar aplicativos de streaming:

   • Em Escolha um método para configurar o aplicativo de processamento de fluxo, escolha Criar do zero.

   • Para configuração do Apache Flink, versão do aplicativo Flink, escolha Apache Flink 1.19.

   • Para configuração do aplicativo:

     • Em Nome do aplicativo, insira MyApplication.

     • Em Descrição, insira My Python test app.

     • Em Acesso aos recursos do aplicativo, escolha Create/update IAM role kinesis-analytics- MyApplication -us-east-1 com as políticas necessárias.

   • Para o modelo para configurações de aplicativos:

     • Em Modelos, escolha Desenvolvimento.

   • Escolha Criar aplicativo de streaming.

Para editar a IAM política para adicionar permissões de bucket do S3

1. Abra o IAM console em https://console.aws.amazon.com/iam/.

2. Selecione Policies (Políticas). Selecione a política kinesis-analytics-service-MyApplication-us-east-1 que o console criou para você na seção anterior.

3. Escolha Editar e, em seguida, escolha a JSONguia.

4. Adicione a seção destacada do exemplo de política a seguir à política. Substitua a conta de amostra IDs (012345678901) com o ID da sua conta.

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:s3:::my-bucket/kinesis-analytics-placeholder-s3-object"
               ]
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-east-1:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-east-1:012345678901:stream/ExampleOutputStream"
           }
       ]
   }

5. Escolha Próximo e, em seguida, escolha Salvar alterações.

Configurar o aplicativo

1. Na MyApplicationpágina, escolha Configurar.

2. Na seção Localização do código do aplicativo:

   • Para o bucket do Amazon S3, selecione o bucket que você criou anteriormente para o código do aplicativo. Escolha Procurar e selecione o bucket correto e, em seguida, escolha Escolher. Não selecione o nome do bucket.

   • Em Caminho do objeto do Amazon S3, insira managed-flink-pyflink-getting-started-1.0.0.zip.

3. Para permissões de acesso, escolha Criar/atualizar IAM função kinesis-analytics-MyApplication-us-east-1 com as políticas necessárias.

4. Vá para as propriedades de tempo de execução e mantenha os valores padrão para todas as outras configurações.

5. Escolha Adicionar novo item e adicione cada um dos seguintes parâmetros:

   ID do grupo	Chave	Valor
   InputStream0	stream.name	ExampleInputStream
   InputStream0	flink.stream.initpos	LATEST
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1
   kinesis.analytics.flink.run.options	python	main.py
   kinesis.analytics.flink.run.options	jarfile	lib/pyflink-dependencies.jar

6. Não modifique nenhuma das outras seções e escolha Salvar alterações.

Executar o aplicativo

1. No console do Amazon Managed Service para Apache Flink, escolha Meu aplicativo e escolha Executar.

2. Na próxima página, na página de configuração de restauração do aplicativo, escolha Executar com o snapshot mais recente e, em seguida, escolha Executar.

   O status nos detalhes do aplicativo muda de Ready para Starting e depois para Running quando o aplicativo é iniciado.

Para abrir o painel do

1. Escolha Abrir painel do Apache Flink. O painel é aberto em uma nova página.

2. Na lista de trabalhos em execução, escolha o único trabalho que você pode ver.

   nota

   Se você definiu as propriedades do Runtime ou editou as IAM políticas incorretamente, o status do aplicativo pode se transformar emRunning, mas o painel do Flink mostra que o trabalho está sendo reiniciado continuamente. Esse é um cenário de falha comum se o aplicativo estiver configurado incorretamente ou não tiver permissões para acessar os recursos externos.

   Quando isso acontecer, verifique a guia Exceções no painel do Flink para ver a causa do problema.

Para ver as métricas

1. Ao lado do botão Atualizar, selecione 10 segundos na lista suspensa.

2. Quando o aplicativo está em execução e em bom estado, você pode ver a métrica de tempo de atividade aumentando continuamente.

3. A métrica de reinicializações completas deve ser zero. Se estiver aumentando, a configuração pode ter problemas. Para investigar o problema, revise a guia Exceções no painel do Flink.

4. A métrica Número de pontos de verificação com falha deve ser zero em um aplicativo saudável.

   nota

   Esse painel exibe um conjunto fixo de métricas com uma granularidade de 5 minutos. Você pode criar um painel de aplicativos personalizado com qualquer métrica no CloudWatch painel.

Para ver a saída

1. Abra o console do Kinesis em https://console.aws.amazon.com /kinesis.

2. Verifique se a região é a mesma que você está usando para executar este tutorial. Por padrão, é US-East-1US East (Norte da Virgínia). Altere a região, se necessário.

3. Escolha fluxos de dados.

4. Selecione o stream que você deseja observar. Para este tutorial, use ExampleOutputStream.

5. Escolha a guia Visualizador de dados.

6. Selecione qualquer fragmento, mantenha Último como posição inicial e escolha Obter registros. Talvez você veja o erro “nenhum registro encontrado para esta solicitação”. Em caso afirmativo, escolha Tentar obter registros novamente. Os registros mais recentes publicados no stream são exibidos.

7. Selecione o valor na coluna Dados para inspecionar o conteúdo do registro em JSON formato.

Como interromper o aplicativo

1. Na lista suspensa Ação, escolha Parar.

2. O status nos detalhes do aplicativo muda de Running para Stopping e depois para Ready quando o aplicativo é completamente interrompido.

   nota

   Não se esqueça também de parar de enviar dados para o stream de entrada a partir do script Python ou do Kinesis Data Generator.