Cree y ejecute un servicio gestionado para la aplicación Apache Flink

Cómo crear flujos de datos (AWS CLI)

1. Para crear la primera transmisión (ExampleInputStream), utilice el siguiente comando de Amazon Kinesis create-stream AWS CLI :

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

2. Para crear la segunda transmisión que la aplicación utiliza para escribir el resultado, ejecute el mismo comando y cambie el nombre de la transmisión aExampleOutputStream:

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

1. Clone el repositorio remoto con el siguiente comando:

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

2. Vaya al directorio amazon-managed-service-for-apache-flink-examples/tree/main/java/GettingStarted.

Para observar los registros

1. Abra la consola de Kinesis en https://console.aws.amazon.com /kinesis.

2. Compruebe que la región es la misma en la que está ejecutando este tutorial, que es us-east-1 US East (Virginia del Norte) de forma predeterminada. Cambie la región si no coincide.

3. Elija Data Streams.

4. Seleccione la transmisión que desee observar, ExampleInputStream o ExampleOutputStream.

5. Seleccione la pestaña Visor de datos.

6. Elija cualquier fragmento, mantenga el valor Último como posición inicial y, a continuación, seleccione Obtener registros. Es posible que aparezca el error «No se ha encontrado ningún registro para esta solicitud». Si es así, selecciona Volver a intentar obtener los registros. Se muestran los registros más recientes publicados en la transmisión.

7. Elija el valor de la columna Datos para inspeccionar el contenido del registro en su JSON formato.

Para cargar el JAR archivo de códigos de la aplicación

1. Abra la consola Amazon S3 en https://console.aws.amazon.com/s3/.

2. Elija el bucket que creó anteriormente para el código de la aplicación.

3. Seleccione Cargar.

4. Elija Add files.

5. Navegue hasta el JAR archivo generado en el paso anterior:target/amazon-msf-java-stream-app-1.0.jar.

6. Elija Cargar sin cambiar ninguna otra configuración.

Para crear la aplicación

1. Abra la consola de Managed Service for Apache Flink en https://console.aws.amazon.com /flink

2. Compruebe que ha seleccionado la región correcta: us-east-1 US East (Norte de Virginia)

3. Abre el menú de la derecha y selecciona Aplicaciones Apache Flink y luego Crear aplicación de streaming. También puede elegir Crear aplicación de streaming en el contenedor Comenzar de la página inicial.

4. En la página Crear aplicación de streaming:

   • Elija un método para configurar la aplicación de procesamiento de transmisiones: elija Crear desde cero.

   • Configuración de Apache Flink, versión de Application Flink: elija Apache Flink 1.19.

5. Configure su aplicación

   • Nombre de la aplicación: introduzcaMyApplication.

   • Descripción: introduzcaMy java test app.

   • Acceso a los recursos de la aplicación: elija Crear o actualizar IAM el rol kinesis-analytics-MyApplication-us-east-1 con las políticas requeridas.

6. Configure su plantilla para los ajustes de la aplicación

   • Plantillas: elija Desarrollo.

7. Selecciona Crear aplicación de streaming en la parte inferior de la página.

Para editar la política

1. Abre la IAM consola en https://console.aws.amazon.com/iam/.

2. Elija Políticas. Elija la política kinesis-analytics-service-MyApplication-us-east-1 que la consola creó en su nombre en la sección anterior.

3. Seleccione Editar y, a continuación, elija la JSONpestaña.

4. Añada la sección subrayada de la siguiente política de ejemplo a la política. Sustituya la cuenta de muestra IDs (012345678901) por tu ID de cuenta.

   {
       "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. Selecciona Siguiente en la parte inferior de la página y, a continuación, selecciona Guardar cambios.

Para editar la configuración

1. En la MyApplicationpágina, elija Configurar.

2. En la sección de ubicación del código de la aplicación:

   • Para el bucket de Amazon S3, seleccione el bucket que creó anteriormente para el código de la aplicación. Elija Examinar y seleccione el bucket correcto y, a continuación, seleccione Elegir. No haga clic en el nombre del depósito.

   • En Ruta al objeto de Amazon S3, introduzca amazon-msf-java-stream-app-1.0.jar.

3. Para los permisos de acceso, selecciona Crear o actualizar IAM el rol kinesis-analytics-MyApplication-us-east-1 con las políticas requeridas.

4. En la sección Propiedades del tiempo de ejecución, añada las siguientes propiedades.

5. Seleccione Añadir nuevo elemento y añada cada uno de los siguientes parámetros:

   ID de grupo	Clave	Valor
   InputStream0	stream.name	ExampleInputStream
   InputStream0	aws.region	us-east-1
   OutputStream0	stream.name	ExampleOutputStream
   OutputStream0	aws.region	us-east-1

6. No modifique ninguna de las demás secciones.

7. Elija Guardar cambios.

Cómo ejecutar la aplicación

1. En la consola de Amazon Managed Service for Apache Flink, seleccione Mi aplicación y, a continuación, Ejecutar.

2. En la página siguiente, la página de configuración de restauración de la aplicación, seleccione Ejecutar con la última instantánea y, a continuación, seleccione Ejecutar.

   El estado de la aplicación detalla las transiciones desde Starting y Ready hasta Running cuando se ha iniciado la aplicación.

Para abrir el panel de

1. Seleccione Abrir el panel de control de Apache Flink. El panel se abre en una página nueva.

2. En la lista de trabajos en ejecución, elige el único trabajo que puedas ver.

   nota

   Si configuras las propiedades de Runtime o editas las IAM políticas de forma incorrecta, el estado de la solicitud podría cambiar aRunning, pero el panel de control de Flink muestra que el trabajo se reinicia continuamente. Este es un escenario de error común si la aplicación está mal configurada o carece de permisos para acceder a los recursos externos.

   Cuando esto suceda, consulte la pestaña Excepciones en el panel de control de Flink para ver la causa del problema.

Para ver las métricas

1. Junto al botón Actualizar, selecciona 10 segundos en la lista desplegable.

2. Cuando la aplicación está en ejecución y en buen estado, puede ver que la métrica de tiempo de actividad aumenta continuamente.

3. La métrica de reinicios completos debe ser cero. Si aumenta, es posible que la configuración tenga problemas. Para investigar el problema, consulta la pestaña Excepciones del panel de control de Flink.

4. La métrica Número de puntos de control fallidos debe ser cero en una aplicación en buen estado.

   nota

   Este panel muestra un conjunto fijo de métricas con una granularidad de 5 minutos. Puede crear un panel de aplicaciones personalizado con cualquier métrica del CloudWatch panel.

Para ver el resultado

1. Abra la consola de Kinesis en https://console.aws.amazon.com /kinesis.

2. Compruebe que la región es la misma que la que está utilizando para ejecutar este tutorial. De forma predeterminada, es US-East-1US East (Virginia del Norte). Cambie la región si es necesario.

3. Elija Flujos de datos.

4. Seleccione la transmisión que desee observar. Para este tutorial, escriba ExampleOutputStream.

5. Seleccione la pestaña Visor de datos.

6. Seleccione cualquier fragmento, mantenga el valor Último como posición inicial y, a continuación, elija Obtener registros. Es posible que aparezca el error «no se ha encontrado ningún registro para esta solicitud». Si es así, selecciona Volver a intentar obtener los registros. Se muestran los registros más recientes publicados en la transmisión.

7. Seleccione el valor en la columna Datos para inspeccionar el contenido del registro en JSON formato.

Cómo detener la aplicación

1. En la lista desplegable Acción, seleccione Detener.

2. El estado en los detalles de la aplicación pasa de Running a yStopping, después, a Ready cuando la aplicación se detiene por completo.

   nota

   No olvide dejar también de enviar datos al flujo de entrada desde el script de Python o el generador de datos de Kinesis.