Crie e execute um aplicativo Managed Service for Apache Flink para o aplicativo Python

Como criar os fluxos de dados (AWS CLI)

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

   $ 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 Conda ou VirtualEnv, onde você instalou a biblioteca Python Flink, esteja ativo no momento.

2. Verifique se você executou o mvn package pelo 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 script main.py com a seguinte configuração:

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

   2. Use as credenciais AWS para acessar os fluxos 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 do IDE e varia.

3. Depois de configurar o IDE, execute o script Python e use as ferramentas fornecidas pelo seu IDE enquanto o aplicativo estiver em execução.

Para fazer upload do arquivo JAR do código do aplicativo

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

2. Selecione 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 Fazer upload sem alterar nenhuma outra configuração.

1. Faça login no Console de gerenciamento da AWS e abra o console do Amazon MSF em https://console.aws.amazon.com/flink.

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 transmissão. Como alternativa, escolha Criar aplicativo de transmissão na seção Introdução da página inicial.

4. Na página Criar aplicativos de transmissão:

   • 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 Criar/atualizar o perfil do IAM kinesis-analytics-MyApplication-us-east-1 com as políticas necessárias.

   • Para Modelo para configurações de aplicativos:

     • Para Modelos, escolha Desenvolvimento.

   • Escolha Criar aplicativo de transmissão.

Editar a política do IAM para adicionar permissões do bucket do S3

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

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

3. Escolha a guia Edit (Editar) e escolha a guia JSON.

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

   JSON

   {
       "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 página MyApplication, selecione Configure (Configurar).

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

   • Para o bucket do Amazon S3, selecione o bucket criado anteriormente para o código do aplicativo. Escolha Procurar, selecione o bucket correto e, em seguida, selecione 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. Em Permissões de acesso, selecione Criar/atualizar o perfil do IAM kinesis-analytics-MyApplication-us-east-1 com as políticas necessárias.

4. Navegue até as Propriedades do Runtime e mantenha os valores padrão para todas as outras configurações.

5. Selecione 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 for Apache Flink, escolha Meu aplicativo e escolha Executar.

2. Na próxima página, que é a 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. Selecione Abrir painel do Apache Flink. O painel é aberto em uma nova página.

2. Na lista Trabalhos em execução, escolha o único trabalho exibido.

   nota

   Se você definiu as propriedades do Runtime ou editou as políticas do IAM incorretamente, o status do aplicativo pode se transformar em Running, 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 recursos externos.

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

Para visualizar 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 íntegro, é possível ver a métrica de tempo de atividade aumentando continuamente.

3. A métrica fullrestarts 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 íntegro.

   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 painel do CloudWatch.

Para visualizar a saída

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

2. Verifique se a região é a mesma utilizada para executar este tutorial. Por padrão, é us-east-1 Leste dos EUA (Norte da Virgínia). Altere a região se necessário.

3. Escolha Fluxos de dados.

4. Selecione o fluxo que deseja observar. Para este tutorial, use ExampleOutputStream.

5. Selecione a guia Visualizador de dados.

6. Selecione qualquer Fragmento, mantenha Último como Posição inicial e, em seguida, 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 fluxo são exibidos.

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

Como interromper o aplicativo

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

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 fluxo de entrada a partir do script Python ou do Kinesis Data Generator.