Crear y ejecutar una aplicación de Managed Service para Apache Flink

Cómo crear flujos de datos (AWS CLI)

1. Para crear el primer flujo (ExampleInputStream), utilice el siguiente comando de la AWS CLI create-stream de Amazon Kinesis.

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

2. Para crear la segunda secuencia que la aplicación utilizará para escribir la salida, ejecute el mismo comando, cambiando el nombre a ExampleOutputStream:

   $ 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.

Observación de 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 Este de EE. UU. (Norte de Virginia) de forma predeterminada. Cambie la región si no coincide.

3. Elija Flujos de datos.

4. Selección del flujo que desee observar, ExampleInputStream o ExampleOutputStream.

5. Seleccione la pestaña Visor de datos.

6. Elija cualquier partición, 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í, seleccione Volver a intentar obtener los registros. Se muestran los registros más recientes publicados en el flujo.

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

Cómo cargar el archivo JAR del código de la aplicación

1. Abra la consola de 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 archivo JAR 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. Inicie sesión en la Consola de administración de AWS y abra la consola de Amazon MSF en https://console.aws.amazon.com/flink.

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

3. Abra el menú de la derecha y seleccione Aplicaciones de Apache Flink y luego Crear aplicación de streaming. También puede elegir Crear aplicación de streaming en el contenedor Introducción 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.20.

5. Configuración de la aplicación

   • Nombre de aplicación: escriba MyApplication.

   • Descripción: escriba My java test app

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

6. Configuración de la plantilla para los ajustes de la aplicación

   • Plantillas: elija Desarrollo.

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

Edición de la política

1. Abra la consola de IAM 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. Elija Editar política y, a continuación, elija la pestaña JSON.

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

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

Edición de la configuración

1. En la página MyApplication, elija Configurar.

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

   • En 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 bucket.

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

3. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM 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. Seleccione Save changes (Guardar cambios).

Cómo ejecutar la aplicación

1. En la consola de Amazon Managed Service para 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 en Detalles de la aplicación cambia de Ready a Starting y luego a 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 nueva página.

2. En la lista Trabajos en ejecución, elija el único trabajo que pueda ver.

   nota

   Si se configuran las propiedades de Runtime o se editan las políticas de IAM de forma incorrecta, el estado de la solicitud podría cambiar a Running, 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.

Visualización de las métricas

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

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

3. La métrica fullrestarts debe ser cero. Si aumenta, es posible que la configuración tenga problemas. Para investigar el problema, consulte 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

   En este panel se muestra un conjunto fijo de métricas con una granularidad de 5 minutos. Se puede crear un panel de aplicaciones personalizado con cualquier métrica del panel de CloudWatch.

Visualización del 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. Por defecto, es us-east-1 Este de EE. UU. (Norte de Virginia). De ser necesario, cambie la región.

3. Elija Flujos de datos.

4. Seleccione el flujo que desea observar. Para este tutorial, escriba ExampleOutputStream.

5. Seleccione la pestaña Visor de datos.

6. Seleccione cualquier partición, 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í, seleccione Volver a intentar obtener los registros. Se muestran los registros más recientes publicados en el flujo.

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

Cómo detener la aplicación

1. En la lista desplegable Acciones, seleccione Detener.

2. El estado en los detalles de la aplicación cambia de Running a Stopping, y después a Ready cuando se ha detenido totalmente la aplicación.

   nota

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