Creación de recursos dependientes Configuración de su entorno de desarrollo local Descarga y análisis del código de Python de flujo de Apache Flink Administración de las dependencias JAR Escritura de registros de muestra en el flujo de entrada Ejecución de la aplicación a nivel local Observación de los datos de entrada y salida en las transmisiones de Kinesis Detención de la ejecución de la aplicación de forma local Empaquetado de su código de la aplicación Cargue el paquete de la aplicación en un bucket de Amazon S3.Creación y ejecución de la aplicación de Managed Service para Apache Flink Siguiente paso

Creación y ejecución de una aplicación de Managed Service para Apache Flink para Python

En esta sección, se creará una aplicación de Managed Service para Apache Flink para una aplicación Python con un flujo de Kinesis como origen y receptor.

Esta sección contiene los siguientes pasos.

Creación de recursos dependientes
Configuración de su entorno de desarrollo local
Descarga y análisis del código de Python de flujo de Apache Flink
Administración de las dependencias JAR
Escritura de registros de muestra en el flujo de entrada
Ejecución de la aplicación a nivel local
Observación de los datos de entrada y salida en las transmisiones de Kinesis
Detención de la ejecución de la aplicación de forma local
Empaquetado de su código de la aplicación
Cargue el paquete de la aplicación en un bucket de Amazon S3.
Creación y ejecución de la aplicación de Managed Service para Apache Flink
Siguiente paso

Creación de recursos dependientes

Antes de crear un Managed Service para Apache Flink para este ejercicio, debe crear los siguientes recursos dependientes:

Dos flujos de Kinesis para entrada y salida.
Un bucket de Amazon S3 para almacenar el código de la aplicación.

nota

En este tutorial se supone que se está implementando su aplicación en la región us-east-1. Si se utiliza otra región, se deben adaptar todos los pasos en consecuencia.

Creación de dos flujos de Kinesis

Antes de crear una aplicación de Managed Service para Apache Flink para este ejercicio, cree dos flujos de datos de Kinesis (ExampleInputStream y ExampleOutputStream) en la misma región que utilizará para implementar la aplicación (us-east-1 en este ejemplo). Su aplicación utiliza estos flujos para los flujos de origen y destino de la aplicación.

Se puede crear estos flujos mediante la consola de Amazon Kinesis o el siguiente comando de la AWS CLI. Para obtener instrucciones sobre la consola, consulte Creating and Updating Data Streams en la Guía para desarrolladores de Amazon Kinesis Data Streams.

Cómo crear flujos de datos (AWS CLI)

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 
```
Para crear el segundo flujo 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 
```

Crear un bucket de Amazon S3

Se puede crear el bucket de Amazon S3 usando la consola. Si desea obtener instrucciones para crear este recurso, consulte los siguientes temas:

¿Cómo se puede crear un bucket de S3? en la Guía de usuario de Amazon Simple Storage Service. Asigne al bucket de Amazon S3 un nombre único globalmente añadiendo su nombre de inicio de sesión.

nota
S3 Asegúrese de crear el bucket en la región que utilice para este tutorial (us-east-1).

Otros recursos

Al crear la aplicación, Managed Service para Apache Flink crea los siguientes recursos de Amazon CloudWatch si aún no existen:

Un grupo de registro llamado /AWS/KinesisAnalytics-java/<my-application>.
Un flujo de registro llamado kinesis-analytics-log-stream.

Configuración de su entorno de desarrollo local

Para el desarrollo y la depuración, se puede ejecutar la aplicación Python Flink en su máquina. Se puede iniciar la aplicación desde la línea de comandos con python main.py o en un IDE de Python de su elección.

nota

En su máquina de desarrollo, debe tener instalados Python 3.10 o 3.11, Java 11, Apache Maven y Git. Se recomienda que utilice un IDE como PyCharm o Visual Studio Code. Para comprobar que cumple todos los requisitos previos, consulte Cumplimiento de los requisitos previos para realizar los ejercicios antes de continuar.

Instalación de la biblioteca PyFlink

Para desarrollar la aplicación y ejecutarla localmente, debe instalar la biblioteca Python de Flink.

Cree un entorno Python independiente con VirtualEnv, Conda o cualquier herramienta similar de Python.
Instale la biblioteca PyFlink en ese entorno. Utilice la misma versión de tiempo de ejecución de Apache Flink que utilizará en Amazon Managed Service para Apache Flink. Actualmente, el tiempo de ejecución recomendado es 1.19.1.
```
$ pip install apache-flink==1.19.1
```
Asegúrese de que el entorn esté activo cuando ejecute la aplicación. Si ejecuta la aplicación en el IDE, asegúrese de que este utilice el entorno como tiempo de ejecución. El procedimiento depende del IDE que esté utilizando.

nota
Solo necesita instalar la biblioteca PyFlink. No se necesita instalar un clúster de Apache Flink en su máquina.

Autenticación de la sesión de AWS

La aplicación utiliza los flujos de datos de Kinesis para publicar datos. Si se ejecuta de forma local, se debe tener una sesión de AWS autenticada válida con permisos para escribir en el flujo de datos de Kinesis. Siga los pasos siguientes para autenticar su sesión:

Si no tiene configurado el perfil con un nombre de AWS CLI y una credencial válida, consulte Cómo configurar la AWS Command Line Interface (AWS CLI).
Compruebe que AWS CLI está correctamente configurado y que sus usuarios tienen permisos para escribir en el flujo de datos de Kinesis publicando el siguiente registro de prueba:
```
$ aws kinesis put-record --stream-name ExampleOutputStream --data TEST --partition-key TEST
```
Si su IDE tiene un complemento con el que integrarse con AWS, puede usarlo para pasar las credenciales a la aplicación que se ejecuta en el IDE. Para obtener más información, consulte AWS Toolkit for PyCharm, AWS Toolkit for Visual Studio Code y AWS Toolkit for IntelliJ IDEA.

Descarga y análisis del código de Python de flujo de Apache Flink

El código de la aplicación de Python para este ejemplo está disponible en GitHub. Para descargar el código de la aplicación, haga lo siguiente:

Clone el repositorio remoto con el siguiente comando:


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

Vaya al directorio ./python/GettingStarted.

Revisión de los componentes de la aplicación

El código de la aplicación se encuentra en main.py. Usamos SQL incrustado en Python para definir el flujo de la aplicación.

nota

Para una experiencia de desarrollador optimizada, la aplicación está diseñada para ejecutarse sin cambios de código tanto en Amazon Managed Service para Apache Flink como de forma local, para el desarrollo en su máquina. La aplicación utiliza la variable de entorno IS_LOCAL = true para detectar si se está ejecutando de forma local. Se debe configurar la variable de entorno IS_LOCAL = true en el intérprete de comandos o en la configuración de ejecución del IDE.

La aplicación configura el entorno de ejecución y lee la configuración del tiempo de ejecución. Para funcionar tanto en Amazon Managed Service para Apache Flink como de forma local, la aplicación comprueba la variable IS_LOCAL.
- El siguiente es el comportamiento predeterminado cuando la aplicación se ejecuta en Amazon Managed Service para Apache Flink:
  1. Cargue las dependencias empaquetadas con la aplicación. Para más información, consulte (enlace)
  2. Cargue la configuración desde las propiedades de tiempo de ejecución que se definan en la aplicación Amazon Managed Service para Apache Flink. Para más información, consulte (enlace)
- Cuando la aplicación detecte IS_LOCAL = true cuando se ejecuta la aplicación de forma local:
  1. Carga las dependencias externas del proyecto.
  2. Carga la configuración desde el archivo application_properties.json incluido en el proyecto.
    
    ... APPLICATION_PROPERTIES_FILE_PATH = "/etc/flink/application_properties.json" ... is_local = ( True if os.environ.get("IS_LOCAL") else False ) ... if is_local: APPLICATION_PROPERTIES_FILE_PATH = "application_properties.json" CURRENT_DIR = os.path.dirname(os.path.realpath(__file__)) table_env.get_config().get_configuration().set_string( "pipeline.jars", "file:///" + CURRENT_DIR + "/target/pyflink-dependencies.jar", )

La aplicación define una tabla de origen con una declaración CREATE TABLE mediante el conector de Kinesis. En esta tabla se leen los datos del flujo de Kinesis de entrada. La aplicación toma el nombre del flujo, la región y la posición inicial de la configuración del tiempo de ejecución.


table_env.execute_sql(f"""
        CREATE TABLE prices (
                ticker VARCHAR(6),
                price DOUBLE,
                event_time TIMESTAMP(3),
                WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
              )
              PARTITIONED BY (ticker)
              WITH (
                'connector' = 'kinesis',
                'stream' = '{input_stream_name}',
                'aws.region' = '{input_stream_region}',
                'format' = 'json',
                'json.timestamp-format.standard' = 'ISO-8601'
              ) """)

En este ejemplo, la aplicación también define una mesa de recepción mediante el conector de Kinesis. Esta tabla envía los datos al flujo de Kinesis de salida.


table_env.execute_sql(f"""
            CREATE TABLE output (
                ticker VARCHAR(6),
                price DOUBLE,
                event_time TIMESTAMP(3)
              )
              PARTITIONED BY (ticker)
              WITH (
                'connector' = 'kinesis',
                'stream' = '{output_stream_name}',
                'aws.region' = '{output_stream_region}',
                'sink.partitioner-field-delimiter' = ';',
                'sink.batch.max-size' = '100',
                'format' = 'json',
                'json.timestamp-format.standard' = 'ISO-8601'
              )""")

Por último, la aplicación ejecuta un código SQL que INSERT INTO... la tabla de receptor de la tabla de origen. En una aplicación más compleja, es probable que tenga que realizar pasos adicionales para transformar los datos antes de escribirlos en el receptor.
```
table_result = table_env.execute_sql("""INSERT INTO output 
        SELECT ticker, price, event_time FROM prices""")
```
Debe añadir otro paso al final de la función main() para ejecutar la aplicación de manera local:
```
if is_local:
    table_result.wait()
```
Sin esta instrucción, la aplicación finaliza de inmediato cuando se ejecuta de manera local. No se debe ejecutar esta declaración cuando ejecute su aplicación en Amazon Managed Service para Apache Flink.

Administración de las dependencias JAR

Una aplicación PyFlink normalmente requiere uno o más conectores. La aplicación de este tutorial usa el conector de Kinesis. Como Apache Flink se ejecuta en la JVM de Java, los conectores se distribuyen como archivos JAR, independientemente de si implementa la aplicación en Python. Se deben empaquetar estas dependencias con la aplicación cuando se la implemente en Amazon Managed Service para Apache Flink.

En este ejemplo, mostramos cómo usar Apache Maven para recuperar las dependencias y empaquetar la aplicación para que se ejecute en Managed Service para Apache Flink.

nota

Existen formas alternativas de buscar y empaquetar las dependencias. En este ejemplo se muestra un método que funciona correctamente con uno o más conectores. También permite ejecutar la aplicación de forma local, para su desarrollo y en Managed Service para Apache Flink sin necesidad de realizar cambios en el código.

Uso del archivo pom.xml

Apache Maven usa el archivo pom.xml para controlar las dependencias y el empaquetado de las aplicaciones.

Todas las dependencias del JAR se especifican en el archivo pom.xml del bloque <dependencies>...</dependencies>.


<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    ...
    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kinesis</artifactId>
            <version>4.3.0-1.19</version>
        </dependency>
    </dependencies>
    ...

Para encontrar el artefacto y la versión del conector correctos que se van a utilizar, consulte Uso de los conectores de Apache Flink con Managed Service para Apache Flink. Asegúrese de consultar la versión de Apache Flink que está utilizando. En este ejemplo, usaremos el conector de Kinesis. Para Apache Flink 1.19, la versión del conector es 4.3.0-1.19.

nota

Si se utiliza Apache Flink 1.19, no hay ninguna versión del conector publicada específicamente para esta versión. Utilice los conectores publicados para la versión 1.18.

Descarga y empaquetado de dependencias

Use Maven para descargar las dependencias definidas en el archivo pom.xml y empaquetarlas para la aplicación Python Flink.

Vaya al directorio que contiene el proyecto de introducción a Python llamado python/GettingStarted.
Ejecuta el siguiente comando:


$ mvn package

Maven crea un nuevo archivo llamado ./target/pyflink-dependencies.jar. Cuando desarrolla localmente en su máquina, la aplicación Python busca este archivo.

nota

Si se olvida de ejecutar este comando, cuando intentes ejecutar la aplicación, esta fallará con el siguiente error: No se ha encontrado ninguna fábrica para el identificador “kinesis”.

Escritura de registros de muestra en el flujo de entrada

En esta sección, se envían registros de muestra al flujo para que los procese la aplicación. Tiene dos opciones para generar datos de muestra: mediante un script de Python o el generador de datos de Kinesis.

Generación de datos de muestra mediante un script de Python

Se puede utilizar un script de Python para enviar registros de muestra al flujo.

nota

Para ejecutar este script de Python, debe usar Python 3.x y tener instalada la biblioteca AWS SDK para Python (Boto).

Para empezar a enviar datos de prueba al flujo de entrada de Kinesis:

Se descarga el script de Python stock.py del generador de datos desde el repositorio de GitHub del generador de datos.
Ejecute el script stock.py:
```
$ python stock.py
```

Mantenga el script en ejecución mientras completa el resto del tutorial. Ahora se puede ejecutar la aplicación Apache Flink.

Genere datos de muestra con Kinesis Data Generator

Como alternativa a la secuencia de comandos de Python, se puede utilizar el Generador de datos de Kinesis, también disponible en una versión alojada, para enviar datos de muestra al azar al flujo. Kinesis Data Generator se ejecuta en su navegador y no se necesita instalar nada en su máquina.

Para configurar y ejecutar Kinesis Data Generator:

Siga las instrucciones de la documentación de Kinesis Data Generator para configurar el acceso a la herramienta. Se ejecutará una plantilla CloudFormation que configurará un usuario y una contraseña.
Acceda a Kinesis Data Generator a través de la URL generada por la plantilla de CloudFormation. Encontrará la URL en la pestaña Resultados una vez que haya completado la plantilla de CloudFormation.
Configure el generador de datos:
- Región: seleccione la región que está utilizando para este tutorial: us-east-1
- Flujo de flujo/entrega: seleccione el flujo de entrada que utilizará la aplicación: ExampleInputStream
- Registros por segundo: 100
- Plantilla de registro: copie y pegue la siguiente plantilla:
```
{
  "event_time" : "{{date.now("YYYY-MM-DDTkk:mm:ss.SSSSS")}},
  "ticker" : "{{random.arrayElement(
        ["AAPL", "AMZN", "MSFT", "INTC", "TBV"]
    )}}",
  "price" : {{random.number(100)}}          
} 
```
Pruebe la plantilla: elija la plantilla de prueba y compruebe que el registro generado es semejante al siguiente:
```
{ "event_time" : "2024-06-12T15:08:32.04800, "ticker" : "INTC", "price" : 7 }
```
Inicie el generador de datos: elija Seleccionar enviar datos.

Kinesis Data Generator ahora envía datos al ExampleInputStream.

Ejecución de la aplicación a nivel local

Se puede probar la aplicación localmente, ejecutándola desde la línea de comandos con python main.py o desde su IDE.

Para ejecutar la aplicación de forma local, se debe tener instalada la versión correcta de la biblioteca PyFlink, tal y como se describe en la sección anterior. Para más información, consulte (enlace)

nota

Antes de continuar, compruebe que las secuencias de entrada y salida estén disponibles. Consulte Crear dos Amazon Kinesis Data Streams. Además, compruebe que tiene permiso para leer y escribir en ambas secuencias. Consulte Autenticación de la sesión de AWS.

Importación del proyecto de Python en su IDE

Para empezar a trabajar en la aplicación en su IDE, debe importarla como proyecto de Python.

El repositorio que ha clonado contiene varios ejemplos. Cada ejemplo es un proyecto independiente. Para este tutorial, se importa el contenido del subdirectorio ./python/GettingStarted a su IDE.

Importe el código como un proyecto de Python existente.

nota

El proceso exacto para importar un nuevo proyecto de Python varía según el IDE que se esté utilizando.

Compruebe la configuración de la aplicación local

Cuando se ejecuta localmente, la aplicación utiliza la configuración del archivo application_properties.json de la carpeta de recursos del proyecto en ./src/main/resources. Se puede editar este archivo para usar diferentes regiones o nombres de flujo de Kinesis.


[
  {
    "PropertyGroupId": "InputStream0",
    "PropertyMap": {
      "stream.name": "ExampleInputStream",
      "flink.stream.initpos": "LATEST",
      "aws.region": "us-east-1"
    }
  },
  {
    "PropertyGroupId": "OutputStream0",
    "PropertyMap": {
      "stream.name": "ExampleOutputStream",
      "aws.region": "us-east-1"
    }
  }
]

Ejecución de la aplicación Python de manera local

Se puede ejecutar la aplicación localmente, ya sea desde la línea de comandos como un script de Python normal o desde el IDE.

Ejecución de la aplicación en la línea de comandos

Asegúrese de que el entorno Python independiente, como Conda o VirtualEnv, en el que instaló la biblioteca Python Flink, esté activo actualmente.
Asegúrese de haber ejecutado mvn package al menos una vez.
Establezca la variable de entorno IS_LOCAL = true:
```
$ export IS_LOCAL=true
```
Ejecute la aplicación como un script de Python normal.
```
$python main.py
```

Ejecución de la aplicación desde el IDE

Configure su IDE para ejecutar el script main.py con la siguiente configuración:
1. Utilice el entorno Python independiente, como Conda o VirtualEnv, donde se instaló la biblioteca PyFlink.
2. Utilice las credenciales AWS para acceder a los flujos de datos de entrada y salida de Kinesis.
3. Configurar IS_LOCAL = true.
El proceso exacto para establecer la configuración de ejecución depende del IDE y varía.
Una vez configurado el IDE, ejecute el script de Python y utilice las herramientas proporcionadas por el IDE mientras se ejecuta la aplicación.

Inspección de los registros de la aplicación localmente

Cuando se ejecuta de manera local, la aplicación no muestra ningún registro en la consola, aparte de unas pocas líneas que se imprimen y se muestran cuando se inicia la aplicación. PyFlink escribe los registros en un archivo del directorio donde está instalada la biblioteca Python Flink. La aplicación imprime la ubicación de los registros cuando se inicia. También puede ejecutar el comando siguiente para encontrar los registros:


$ python -c "import pyflink;import os;print(os.path.dirname(os.path.abspath(pyflink.__file__))+'/log')"

Enumere los archivos en el directorio de registro. Por lo general, encontrará un único archivo .log.
Siga el archivo mientras se ejecuta la aplicación: tail -f <log-path>/<log-file>.log.

Observación de los datos de entrada y salida en las transmisiones de Kinesis

Se pueden observar los registros enviados al flujo de entrada por el (que genera un ejemplo de Python) o el generador de datos de Kinesis (enlace) mediante el visor de datos de la consola de Amazon Kinesis.

Observación de los registros:

Detención de la ejecución de la aplicación de forma local

Detenga la aplicación que se está ejecutando en el IDE. El IDE normalmente ofrece una opción de “parada”. La ubicación y el método exactos dependen del IDE.

Empaquetado de su código de la aplicación

En esta sección, se usa Apache Maven para empaquetar el código de la aplicación y todas las dependencias necesarias en un archivo .zip.

Vuelva a ejecutar el comando del paquete Maven:


$ mvn package

Este comando genera el archivo target/managed-flink-pyflink-getting-started-1.0.0.zip.

Cargue el paquete de la aplicación en un bucket de Amazon S3.

En esta sección, se cargará el archivo .zip que creó en la sección anterior en el bucket de Amazon Simple Storage Service (Amazon S3) que creó al principio de este tutorial. Si no se ha completado este paso, consulte (enlace).

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

Abra la consola de Amazon S3 en https://console.aws.amazon.com/s3.
Elija el bucket que creó anteriormente para el código de la aplicación.
Seleccione Cargar.
Elija Add files.
Navegue hasta el archivo.zip generado en el paso anterior: target/managed-flink-pyflink-getting-started-1.0.0.zip.
Elija Cargar sin cambiar ninguna otra configuración.

Creación y ejecución de la aplicación de Managed Service para Apache Flink

Se puede crear y ejecutar una aplicación de Managed Service para Apache Flink mediante la consola o la AWS CLI. Para este tutorial, usaremos la consola.

Creación de la aplicación

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.
Compruebe que ha seleccionado la región correcta: Este de EE. UU. (Norte de Virginia) us-east-1.
Abra el menú de la derecha y seleccione Aplicaciones de Apache Flink y luego Crear aplicación de streaming. Como alternativa, seleccione Crear aplicación de flujo en la sección Introducción de la página inicial.
En la página Crear aplicaciones de flujo:
- En Elija un método para configurar la aplicación de procesamiento de transmisiones, elija Crear desde cero.
- En Configuración de Apache Flink, versión Application Flink, elija Apache Flink 1.19.
- En Configuración de aplicaciones:
  - En Nombre de la aplicación, escriba MyApplication.
  - En Descripción, escriba My Python test app.
  - En Acceso a los recursos de la aplicación, elija Crear/actualizar el rol de IAM kinesis-analytics-MyApplication-us-east-1 con las políticas requeridas.
- En Configuración de la plantilla de aplicaciones:
  - En Plantillas, elija Desarrollo.
- Elija Crear aplicación de flujos.

nota

Al crear una aplicación de Managed Service para Apache Flink mediante la consola, tiene la opción de tener un rol de IAM y una política creada para su aplicación. La aplicación utiliza este rol y la política para acceder a los recursos dependientes. Estos recursos de IAM reciben un nombre usando el nombre de la aplicación y la región tal y como se indica a continuación:

Política: kinesis-analytics-service-MyApplication-us-west-2
Rol: : kinesisanalytics-MyApplication-us-west-2

Amazon Managed Service para Apache Flink Amazon se denominaba anteriormente Kinesis Data Analytics. El nombre de los recursos que se generan automáticamente lleva el prefijo kinesis-analytics por motivos de compatibilidad con versiones anteriores.

Modificar la política de IAM

Edite la política de IAM para añadir los permisos para acceder al bucket de Amazon S3.

Cómo editar la política de IAM para añadir los permisos para el bucket de S3

Abra la consola de IAM en https://console.aws.amazon.com/iam/.
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.
Elija Editar política y, a continuación, elija la pestaña JSON.

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.

Elija Guardar cambios y después Probar.

Configurar la aplicación

Edite la configuración de la aplicación para establecer el artefacto del código de la aplicación.

Cómo configurar la aplicación

En la página MyApplication, elija Configurar.
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 seleccione el nombre del bucket.
- En Ruta al objeto de Amazon S3, introduzca managed-flink-pyflink-getting-started-1.0.0.zip.
En Permisos de acceso, seleccione Crear/actualizar kinesis-analytics-MyApplication-us-east-1 del rol de IAM con las políticas requeridas.
Vaya a Propiedades de tiempo de ejecución y mantenga los valores predeterminados para todas las demás configuraciones.

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`	`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`

No modifique ninguna de las demás secciones y elija Guardar cambios.

nota

Al activar el registro de Amazon CloudWatch, Managed Service para Apache Flink crea un grupo de registro y un flujo de registro. Los nombres de estos recursos son los siguientes:

Grupo de registro: /aws/kinesis-analytics/MyApplication
Flujo de registro: kinesis-analytics-log-stream

Ejecución de la aplicación

La aplicación ya está configurada y lista para ejecutarse.

Cómo ejecutar la aplicación

En la consola de Amazon Managed Service para Apache Flink, seleccione Mi aplicación y, a continuación, Ejecutar.
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.

Cuando la aplicación esté en el estado Running, se puede abrir el panel de control de Flink.

Para abrir el panel de

Seleccione Abrir el panel de control de Apache Flink. El panel se abre en una nueva página.
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.

Observación de las métricas de la aplicación en ejecución

En la página myApplication, en la sección de métricas de Amazon CloudWatch, se pueden ver algunas de las métricas fundamentales de la aplicación en ejecución.

Visualización de las métricas

Junto al botón Actualizar, seleccione 10 segundos en la lista desplegable.
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.
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.
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.

Observación de los datos de salida en los flujos de Kinesis

Asegúrese de seguir publicando datos en la entrada, ya sea mediante el script de Python o el generador de datos de Kinesis.

Ahora puede observar el resultado de la aplicación que se ejecuta en Managed Service para Apache Flink mediante el visor de datos de https://console.aws.amazon.com/kinesis/, de forma similar a lo que hacía anteriormente.

Visualización del resultado

Abra la consola de Kinesis en https://console.aws.amazon.com/kinesis.
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.
Elija Flujos de datos.
Seleccione el flujo que desea observar. Para este tutorial, escriba ExampleOutputStream.
Seleccione la pestaña Visor de datos.
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.
Seleccione el valor en la columna Datos para inspeccionar el contenido del registro en formato JSON.

Detener la aplicación

Para detener la aplicación, vaya a la página de la consola de la aplicación de Managed Service para Apache Flink denominada MyApplication.

Cómo detener la aplicación

En la lista desplegable Acciones, seleccione Detener.
El estado en 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.

Siguiente paso

Limpieza de recursos de AWS

Aviso JavaScript está desactivado o no está disponible en su navegador.

Para utilizar la documentación de AWS, debe estar habilitado JavaScript. Para obtener más información, consulte las páginas de ayuda de su navegador.

Convenciones del documento

Tutorial: introducción al uso de Python en Managed Service para Apache Flink

Eliminar recursos