Información de versiones anteriores de Managed Service para Apache Flink

1. Asegúrese de tener instalado Apache Maven y de que su variable de entorno JAVA_HOME apunte a un JDK en lugar de un JRE. Se puede probar la instalación de Apache Maven con el siguiente comando:

   mvn -version

2. Descargue el código fuente de la versión 1.8.2 de Apache Flink:

   wget https://archive.apache.org/dist/flink/flink-1.8.2/flink-1.8.2-src.tgz

3. Descomprima el código fuente de Apache Flink:

   tar -xvf flink-1.8.2-src.tgz

4. Cambie al directorio del código fuente de Apache Flink:

   cd flink-1.8.2

5. Compile e instale Apache Flink:

   mvn clean install -Pinclude-kinesis -DskipTests

   nota

   Si está compilando Flink en Microsoft Windows, debe agregar el parámetro -Drat.skip=true.

Procedimiento para registrarse en Cuenta de AWS

1. Abra https://portal.aws.amazon.com/billing/signup.

2. Siga las instrucciones que se le indiquen.

   Parte del procedimiento de registro consiste en recibir una llamada telefónica o mensaje de texto e indicar un código de verificación en el teclado del teléfono.

   Al registrarse en una Cuenta de AWS, se crea un Usuario raíz de la cuenta de AWS. El usuario raíz tendrá acceso a todos los Servicios de AWS y recursos de esa cuenta. Como práctica recomendada de seguridad, asigne acceso administrativo a un usuario y utilice únicamente el usuario raíz para realizar tareas que requieren acceso de usuario raíz.

Protección de Usuario raíz de la cuenta de AWS

1. Inicie sesión en Consola de administración de AWS como propietario de la cuenta; para ello, elija Usuario raíz e introduzca el correo electrónico de su Cuenta de AWS. En la siguiente página, escriba su contraseña.

   Para obtener ayuda para iniciar sesión con el usuario raíz, consulte Iniciar sesión como usuario raíz en la Guía del usuario de AWS Sign-In.

2. Active la autenticación multifactor (MFA) para el usuario raíz.

   Para obtener instrucciones, consulte Habilitación de un dispositivo MFA virtual para su usuario raíz de la Cuenta de AWS (consola) en la Guía del usuario de IAM.

Creación de un usuario con acceso administrativo

1. Activar IAM Identity Center.

   Consulte las instrucciones en Activar AWS IAM Identity Center en la Guía del usuario de AWS IAM Identity Center.

2. En IAM Identity Center, conceda acceso administrativo a un usuario.

   Para ver un tutorial sobre cómo usar Directorio de IAM Identity Center como origen de identidad, consulte Configuración del acceso de los usuarios con el Directorio de IAM Identity Center predeterminado en la Guía del usuario de AWS IAM Identity Center.

Inicio de sesión como usuario con acceso de administrador

• Para iniciar sesión con el usuario de IAM Identity Center, use la URL de inicio de sesión que se envió a la dirección de correo electrónico cuando creó el usuario de IAM Identity Center.

  Para obtener ayuda para iniciar sesión con un usuario de IAM Identity Center, consulte Inicio de sesión en el portal de acceso de AWS en la Guía del usuario de AWS Sign-In.

Concesión de acceso a usuarios adicionales

1. En IAM Identity Center, cree un conjunto de permisos que siga la práctica recomendada de aplicar permisos de privilegios mínimos.

   Para conocer las instrucciones, consulte Create a permission set en la Guía del usuario de AWS IAM Identity Center.

2. Asigne usuarios a un grupo y, a continuación, asigne el acceso de inicio de sesión único al grupo.

   Para conocer las instrucciones, consulte Add groups en la Guía del usuario de AWS IAM Identity Center.

Para configurar la AWS CLI

1. Descargue y configure la AWS CLI. Para obtener instrucciones, consulte los siguientes temas en la Guía del usuario de la AWS Command Line Interface:

   • Instalación del AWS Command Line Interface

   • Configuración de la AWS CLI

2. Añada un perfil con nombre para el usuario administrador en el archivo config de la AWS CLI. Se puede utilizar este perfil cuando ejecute los comandos de la AWS CLI. Para obtener más información sobre los perfiles con nombre, consulte Perfiles con nombre en la Guía del usuario de AWS Command Line Interface.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   Para ver una lista de las regiones de AWS disponibles, consulte Regions and Endpoints en la Referencia general de Amazon Web Services.

   nota

   El código y los comandos de ejemplo de este tutorial utilizan la región Oeste de EE. UU. (Oregón). Para usar una región diferente, cambie la región en el código y los comandos de este tutorial por la región que desea usar.

3. Verifique la configuración introduciendo el siguiente comando de ayuda en el símbolo del sistema:

   aws help

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-west-2 \
   --profile adminuser

2. 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-west-2 \
   --profile adminuser

1. Cree un archivo denominado stock.py con el siguiente contenido:

   
       import datetime
       import json
       import random
       import boto3
       STREAM_NAME = "ExampleInputStream"
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2')) 

2. Más adelante en el tutorial, se ejecuta el script stock.py para enviar datos a la aplicación.

   $ python stock.py

1. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. Vaya al directorio amazon-kinesis-data-analytics-java-examples/GettingStarted.

Cómo compilar el código de la aplicación

1. Para utilizar el código de la aplicación, compile y empaquete el código en un archivo JAR. Se puede compilar y empaquetar el código de una de las dos formas siguientes:

   • Utilice la herramienta de línea de comandos de Maven. Cree su archivo JAR ejecutando el siguiente comando en el directorio que contiene el archivo pom.xml:

     mvn package -Dflink.version=1.13.2

   • Use el entorno de desarrollo. Consulte la documentación de su entorno de desarrollo para obtener más información.

     nota

     El código fuente proporcionado se basa en bibliotecas de Java 11.

   Se puede cargar el paquete como un archivo JAR o puede comprimir el paquete y cargarlo como un archivo ZIP. Si crea su aplicación mediante la AWS CLI, debe especificar el tipo de contenido del código (JAR o ZIP).

2. Si hay errores al compilar, verifique que la variable de entorno JAVA_HOME se ha configurado correctamente.

Cómo cargar el código de la aplicación

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

2. Elija Crear bucket.

3. Escriba ka-app-code-<username> en el campo Nombre del bucket. Añada un sufijo al nombre del bucket, como su nombre de usuario, para que sea único a nivel global. Elija Siguiente.

4. En el paso Configurar opciones, deje los ajustes tal y como están y elija Siguiente.

5. En el paso Establecer permisos, deje los ajustes tal y como están y elija Siguiente.

6. Elija Crear bucket.

7. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y seleccione Subir.

8. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior. Elija Siguiente.

9. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Descripción, escriba My java test app.

   • En Tiempo de ejecución, escriba Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.13.

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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:::ka-app-code-username/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca aws-kinesis-analytics-java-apps-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. Introduzca lo siguiente:

   ID de grupo	Clave	Valor
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

5. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

6. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

7. Elija Actualizar.

Cómo crear un rol de IAM

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

2. En el panel de navegación, elija Roles, Crear rol.

3. En Seleccionar tipo de entidad de confianza, elija Servicio de AWS. En Elegir el servicio que usará este rol, elija Kinesis. En Seleccionar su caso de uso, elija Kinesis Analytics.

   Elija Siguiente: permisos.

4. En la página Asociar políticas de permisos, elija Siguiente: Revisión. Asociará políticas de permisos después de crear el rol.

5. En la página Crear rol, escriba MF-stream-rw-role como Nombre de rol. Elija Crear rol.

   Ahora ha creado un nuevo rol de IAM llamado MF-stream-rw-role. A continuación, actualice las políticas de confianza y permisos del rol.

6. Asocie la política de permisos al rol.

   nota

   Para este ejercicio, Managed Service para Apache Flink asume este rol tanto para leer datos de un flujo de datos de Kinesis (origen) como para escribir la salida en otro flujo de datos de Kinesis. Asocie la política que ha creado en el paso anterior, Creación de una política de permisos.

   1. En la página Resumen, elija la pestaña Permisos.

   2. Seleccione Asociar políticas.

   3. En el campo de búsqueda, escriba AKReadSourceStreamWriteSinkStream (la política que ha creado en la sección anterior).

   4. Elija la política AKReadSourceStreamWriteSinkStream y luego elija Asociar política.

1. Guarde el siguiente código JSON en un archivo denominado create_request.json. Cambie el ARN del rol de ejemplo por el ARN del rol que ha creado antes. Reemplace el sufijo del ARN del bucket (username) por el sufijo que eligió en la sección anterior. Reemplace el ID de la cuenta de ejemplo (012345678901) del rol de ejecución del servicio por el ID de su cuenta.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_15",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Ejecute la acción CreateApplication con la solicitud anterior para crear la aplicación:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

Cómo iniciar la aplicación

1. Guarde el siguiente código JSON en un archivo denominado start_request.json.

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. Ejecute la acción StartApplication con la solicitud anterior para iniciar la aplicación:

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

Cómo detener la aplicación

1. Guarde el siguiente código JSON en un archivo denominado stop_request.json.

   {
      "ApplicationName": "test"
   }
   

2. Ejecute la acción StopApplication con la siguiente solicitud para detener la aplicación:

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

Cómo actualizar las propiedades de entorno de la aplicación

1. Guarde el siguiente código JSON en un archivo denominado update_properties_request.json.

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. Ejecute la acción UpdateApplication con la solicitud anterior para actualizar las propiedades del entorno:

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminació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. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

4. En la página Flujos de Kinesis, elija ExampleOutputStream, elija Acciones, elija Eliminar y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

Procedimiento para registrarse en Cuenta de AWS

1. Abra https://portal.aws.amazon.com/billing/signup.

2. Siga las instrucciones que se le indiquen.

   Parte del procedimiento de registro consiste en recibir una llamada telefónica o mensaje de texto e indicar un código de verificación en el teclado del teléfono.

   Al registrarse en una Cuenta de AWS, se crea un Usuario raíz de la cuenta de AWS. El usuario raíz tendrá acceso a todos los Servicios de AWS y recursos de esa cuenta. Como práctica recomendada de seguridad, asigne acceso administrativo a un usuario y utilice únicamente el usuario raíz para realizar tareas que requieren acceso de usuario raíz.

Protección de Usuario raíz de la cuenta de AWS

1. Inicie sesión en Consola de administración de AWS como propietario de la cuenta; para ello, elija Usuario raíz e introduzca el correo electrónico de su Cuenta de AWS. En la siguiente página, escriba su contraseña.

   Para obtener ayuda para iniciar sesión con el usuario raíz, consulte Iniciar sesión como usuario raíz en la Guía del usuario de AWS Sign-In.

2. Active la autenticación multifactor (MFA) para el usuario raíz.

   Para obtener instrucciones, consulte Habilitación de un dispositivo MFA virtual para su usuario raíz de la Cuenta de AWS (consola) en la Guía del usuario de IAM.

Creación de un usuario con acceso administrativo

1. Activar IAM Identity Center.

   Consulte las instrucciones en Activar AWS IAM Identity Center en la Guía del usuario de AWS IAM Identity Center.

2. En IAM Identity Center, conceda acceso administrativo a un usuario.

   Para ver un tutorial sobre cómo usar Directorio de IAM Identity Center como origen de identidad, consulte Configuración del acceso de los usuarios con el Directorio de IAM Identity Center predeterminado en la Guía del usuario de AWS IAM Identity Center.

Inicio de sesión como usuario con acceso de administrador

• Para iniciar sesión con el usuario de IAM Identity Center, use la URL de inicio de sesión que se envió a la dirección de correo electrónico cuando creó el usuario de IAM Identity Center.

  Para obtener ayuda para iniciar sesión con un usuario de IAM Identity Center, consulte Inicio de sesión en el portal de acceso de AWS en la Guía del usuario de AWS Sign-In.

Concesión de acceso a usuarios adicionales

1. En IAM Identity Center, cree un conjunto de permisos que siga la práctica recomendada de aplicar permisos de privilegios mínimos.

   Para conocer las instrucciones, consulte Create a permission set en la Guía del usuario de AWS IAM Identity Center.

2. Asigne usuarios a un grupo y, a continuación, asigne el acceso de inicio de sesión único al grupo.

   Para conocer las instrucciones, consulte Add groups en la Guía del usuario de AWS IAM Identity Center.

Para configurar la AWS CLI

1. Descargue y configure la AWS CLI. Para obtener instrucciones, consulte los siguientes temas en la Guía del usuario de la AWS Command Line Interface:

   • Instalación del AWS Command Line Interface

   • Configuración de la AWS CLI

2. Añada un perfil con nombre para el usuario administrador en el archivo config de la AWS CLI. Se puede utilizar este perfil cuando ejecute los comandos de la AWS CLI. Para obtener más información sobre los perfiles con nombre, consulte Perfiles con nombre en la Guía del usuario de AWS Command Line Interface.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   Para ver una lista de las regiones de AWS disponibles, consulte Regions and Endpoints en la Referencia general de Amazon Web Services.

   nota

   El código y los comandos de ejemplo de este tutorial utilizan la región Oeste de EE. UU. (Oregón). Para usar una región diferente, cambie la región en el código y los comandos de este tutorial por la región que desea usar.

3. Verifique la configuración introduciendo el siguiente comando de ayuda en el símbolo del sistema:

   aws help

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-west-2 \
   --profile adminuser

2. 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-west-2 \
   --profile adminuser

1. Cree un archivo denominado stock.py con el siguiente contenido:

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. Más adelante en el tutorial, se ejecuta el script stock.py para enviar datos a la aplicación.

   $ python stock.py

1. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. Vaya al directorio amazon-kinesis-data-analytics-java-examples/GettingStarted.

Cómo compilar el código de la aplicación

1. Para utilizar el código de la aplicación, compile y empaquete el código en un archivo JAR. Se puede compilar y empaquetar el código de una de las dos formas siguientes:

   • Utilice la herramienta de línea de comandos de Maven. Cree su archivo JAR ejecutando el siguiente comando en el directorio que contiene el archivo pom.xml:

     mvn package -Dflink.version=1.11.3

   • Use el entorno de desarrollo. Consulte la documentación de su entorno de desarrollo para obtener más información.

     nota

     El código fuente proporcionado se basa en bibliotecas de Java 11. Asegúrese de que la versión Java de su proyecto sea 11.

   Se puede cargar el paquete como un archivo JAR o puede comprimir el paquete y cargarlo como un archivo ZIP. Si crea su aplicación mediante la AWS CLI, debe especificar el tipo de contenido del código (JAR o ZIP).

2. Si hay errores al compilar, verifique que la variable de entorno JAVA_HOME se ha configurado correctamente.

Cómo cargar el código de la aplicación

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

2. Elija Crear bucket.

3. Escriba ka-app-code-<username> en el campo Nombre del bucket. Añada un sufijo al nombre del bucket, como su nombre de usuario, para que sea único a nivel global. Elija Siguiente.

4. En el paso Configurar opciones, deje los ajustes tal y como están y elija Siguiente.

5. En el paso Establecer permisos, deje los ajustes tal y como están y elija Siguiente.

6. Elija Crear bucket.

7. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y seleccione Subir.

8. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior. Elija Siguiente.

9. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Descripción, escriba My java test app.

   • En Tiempo de ejecución, escriba Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.11 (versión recomendada).

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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:::ka-app-code-username/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca aws-kinesis-analytics-java-apps-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. En Propiedades, en ID de grupo, escriba ProducerConfigProperties.

5. Escriba las siguientes propiedades y valores de la aplicación:

   ID de grupo	Clave	Valor
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

6. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

7. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

8. Elija Actualizar.

Cómo crear un rol de IAM

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

2. En el panel de navegación, elija Roles, Crear rol.

3. En Seleccionar tipo de entidad de confianza, elija Servicio de AWS. En Elegir el servicio que usará este rol, elija Kinesis. En Seleccionar su caso de uso, elija Kinesis Analytics.

   Elija Siguiente: permisos.

4. En la página Asociar políticas de permisos, elija Siguiente: Revisión. Asociará políticas de permisos después de crear el rol.

5. En la página Crear rol, escriba MF-stream-rw-role como Nombre de rol. Elija Crear rol.

   Ahora ha creado un nuevo rol de IAM llamado MF-stream-rw-role. A continuación, actualice las políticas de confianza y permisos del rol.

6. Asocie la política de permisos al rol.

   nota

   Para este ejercicio, Managed Service para Apache Flink asume este rol tanto para leer datos de un flujo de datos de Kinesis (origen) como para escribir la salida en otro flujo de datos de Kinesis. Asocie la política que ha creado en el paso anterior, Crear una política de permisos.

   1. En la página Resumen, elija la pestaña Permisos.

   2. Seleccione Asociar políticas.

   3. En el campo de búsqueda, escriba AKReadSourceStreamWriteSinkStream (la política que ha creado en la sección anterior).

   4. Elija la política AKReadSourceStreamWriteSinkStream y luego elija Asociar política.

1. Guarde el siguiente código JSON en un archivo denominado create_request.json. Cambie el ARN del rol de ejemplo por el ARN del rol que ha creado antes. Reemplace el sufijo del ARN del bucket (username) por el sufijo que eligió en la sección anterior. Reemplace el ID de la cuenta de ejemplo (012345678901) del rol de ejecución del servicio por el ID de su cuenta.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_11",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Ejecute la acción CreateApplication con la solicitud anterior para crear la aplicación:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

Cómo iniciar la aplicación

1. Guarde el siguiente código JSON en un archivo denominado start_request.json.

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. Ejecute la acción StartApplication con la solicitud anterior para iniciar la aplicación:

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

Cómo detener la aplicación

1. Guarde el siguiente código JSON en un archivo denominado stop_request.json.

   
   {
       "ApplicationName": "test"
   }
   

2. Ejecute la acción StopApplication con la siguiente solicitud para detener la aplicación:

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

Cómo actualizar las propiedades de entorno de la aplicación

1. Guarde el siguiente código JSON en un archivo denominado update_properties_request.json.

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. Ejecute la acción UpdateApplication con la solicitud anterior para actualizar las propiedades del entorno:

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminació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. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

4. En la página Flujos de Kinesis, elija ExampleOutputStream, elija Acciones, elija Eliminar y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

Procedimiento para registrarse en Cuenta de AWS

1. Abra https://portal.aws.amazon.com/billing/signup.

2. Siga las instrucciones que se le indiquen.

   Parte del procedimiento de registro consiste en recibir una llamada telefónica o mensaje de texto e indicar un código de verificación en el teclado del teléfono.

   Al registrarse en una Cuenta de AWS, se crea un Usuario raíz de la cuenta de AWS. El usuario raíz tendrá acceso a todos los Servicios de AWS y recursos de esa cuenta. Como práctica recomendada de seguridad, asigne acceso administrativo a un usuario y utilice únicamente el usuario raíz para realizar tareas que requieren acceso de usuario raíz.

Protección de Usuario raíz de la cuenta de AWS

1. Inicie sesión en Consola de administración de AWS como propietario de la cuenta; para ello, elija Usuario raíz e introduzca el correo electrónico de su Cuenta de AWS. En la siguiente página, escriba su contraseña.

   Para obtener ayuda para iniciar sesión con el usuario raíz, consulte Iniciar sesión como usuario raíz en la Guía del usuario de AWS Sign-In.

2. Active la autenticación multifactor (MFA) para el usuario raíz.

   Para obtener instrucciones, consulte Habilitación de un dispositivo MFA virtual para su usuario raíz de la Cuenta de AWS (consola) en la Guía del usuario de IAM.

Creación de un usuario con acceso administrativo

1. Activar IAM Identity Center.

   Consulte las instrucciones en Activar AWS IAM Identity Center en la Guía del usuario de AWS IAM Identity Center.

2. En IAM Identity Center, conceda acceso administrativo a un usuario.

   Para ver un tutorial sobre cómo usar Directorio de IAM Identity Center como origen de identidad, consulte Configuración del acceso de los usuarios con el Directorio de IAM Identity Center predeterminado en la Guía del usuario de AWS IAM Identity Center.

Inicio de sesión como usuario con acceso de administrador

• Para iniciar sesión con el usuario de IAM Identity Center, use la URL de inicio de sesión que se envió a la dirección de correo electrónico cuando creó el usuario de IAM Identity Center.

  Para obtener ayuda para iniciar sesión con un usuario de IAM Identity Center, consulte Inicio de sesión en el portal de acceso de AWS en la Guía del usuario de AWS Sign-In.

Concesión de acceso a usuarios adicionales

1. En IAM Identity Center, cree un conjunto de permisos que siga la práctica recomendada de aplicar permisos de privilegios mínimos.

   Para conocer las instrucciones, consulte Create a permission set en la Guía del usuario de AWS IAM Identity Center.

2. Asigne usuarios a un grupo y, a continuación, asigne el acceso de inicio de sesión único al grupo.

   Para conocer las instrucciones, consulte Add groups en la Guía del usuario de AWS IAM Identity Center.

Para configurar la AWS CLI

1. Descargue y configure la AWS CLI. Para obtener instrucciones, consulte los siguientes temas en la Guía del usuario de la AWS Command Line Interface:

   • Instalación del AWS Command Line Interface

   • Configuración de la AWS CLI

2. Añada un perfil con nombre para el usuario administrador en el archivo config de la AWS CLI. Se puede utilizar este perfil cuando ejecute los comandos de la AWS CLI. Para obtener más información sobre los perfiles con nombre, consulte Perfiles con nombre en la Guía del usuario de AWS Command Line Interface.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   Para ver una lista de las regiones disponibles, consulte Regiones y puntos de enlace en la Referencia general de Amazon Web Services.

   nota

   El código y los comandos de ejemplo de este tutorial utilizan la región Oeste de EE. UU. (Oregón). Para usar una región de AWS diferente, cambie la región en el código y los comandos de este tutorial por la región que desea usar.

3. Verifique la configuración introduciendo el siguiente comando de ayuda en el símbolo del sistema:

   aws help

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-west-2 \
   --profile adminuser

2. 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-west-2 \
   --profile adminuser

1. Cree un archivo denominado stock.py con el siguiente contenido:

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. Más adelante en el tutorial, se ejecuta el script stock.py para enviar datos a la aplicación.

   $ python stock.py

1. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. Vaya al directorio amazon-kinesis-data-analytics-java-examples/GettingStarted_1_8.

Cómo compilar el código de la aplicación

1. Para utilizar el código de la aplicación, compile y empaquete el código en un archivo JAR. Se puede compilar y empaquetar el código de una de las dos formas siguientes:

   • Utilice la herramienta de línea de comandos de Maven. Cree su archivo JAR ejecutando el siguiente comando en el directorio que contiene el archivo pom.xml:

     mvn package -Dflink.version=1.8.2

   • Use el entorno de desarrollo. Consulte la documentación de su entorno de desarrollo para obtener más información.

     nota

     El código fuente proporcionado se basa en bibliotecas de Java 1.8. Asegúrese de que la versión Java de su proyecto sea 1.8.

   Se puede cargar el paquete como un archivo JAR o puede comprimir el paquete y cargarlo como un archivo ZIP. Si crea su aplicación mediante la AWS CLI, debe especificar el tipo de contenido del código (JAR o ZIP).

2. Si hay errores al compilar, verifique que la variable de entorno JAVA_HOME se ha configurado correctamente.

Cómo cargar el código de la aplicación

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

2. Elija Crear bucket.

3. Escriba ka-app-code-<username> en el campo Nombre del bucket. Añada un sufijo al nombre del bucket, como su nombre de usuario, para que sea único a nivel global. Elija Siguiente.

4. En el paso Configurar opciones, deje los ajustes tal y como están y elija Siguiente.

5. En el paso Establecer permisos, deje los ajustes tal y como están y elija Siguiente.

6. Elija Crear bucket.

7. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y seleccione Subir.

8. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior. Elija Siguiente.

9. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Descripción, escriba My java test app.

   • En Tiempo de ejecución, escriba Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.8 (versión recomendada).

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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:::ka-app-code-username/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca aws-kinesis-analytics-java-apps-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. Escriba las siguientes propiedades y valores de la aplicación:

   ID de grupo	Clave	Valor
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

5. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

6. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

7. Elija Actualizar.

1. En la página Mi aplicación, elija Ejecutar. Confirme la acción.

2. Cuando la aplicación se está ejecutando, actualice la página. La consola muestra el Gráfico de la aplicación.

Cómo crear un rol de IAM

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

2. En el panel de navegación, elija Roles, Crear rol.

3. En Seleccionar tipo de entidad de confianza, elija Servicio de AWS. En Elegir el servicio que usará este rol, elija Kinesis. En Seleccionar su caso de uso, elija Kinesis Analytics.

   Elija Siguiente: permisos.

4. En la página Asociar políticas de permisos, elija Siguiente: Revisión. Asociará políticas de permisos después de crear el rol.

5. En la página Crear rol, escriba MF-stream-rw-role como Nombre de rol. Elija Crear rol.

   Ahora ha creado un nuevo rol de IAM llamado MF-stream-rw-role. A continuación, actualice las políticas de confianza y permisos del rol.

6. Asocie la política de permisos al rol.

   nota

   Para este ejercicio, Managed Service para Apache Flink asume este rol tanto para leer datos de un flujo de datos de Kinesis (origen) como para escribir la salida en otro flujo de datos de Kinesis. Asocie la política que ha creado en el paso anterior, Crear una política de permisos.

   1. En la página Resumen, elija la pestaña Permisos.

   2. Seleccione Asociar políticas.

   3. En el campo de búsqueda, escriba AKReadSourceStreamWriteSinkStream (la política que ha creado en la sección anterior).

   4. Elija la política AKReadSourceStreamWriteSinkStream y luego elija Asociar política.

1. Guarde el siguiente código JSON en un archivo denominado create_request.json. Cambie el ARN del rol de ejemplo por el ARN del rol que ha creado antes. Reemplace el sufijo del ARN del bucket (username) por el sufijo que eligió en la sección anterior. Reemplace el ID de la cuenta de ejemplo (012345678901) del rol de ejecución del servicio por el ID de su cuenta.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_8",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "aws-kinesis-analytics-java-apps-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Ejecute la acción CreateApplication con la solicitud anterior para crear la aplicación:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

Cómo iniciar la aplicación

1. Guarde el siguiente código JSON en un archivo denominado start_request.json.

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. Ejecute la acción StartApplication con la solicitud anterior para iniciar la aplicación:

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

Cómo detener la aplicación

1. Guarde el siguiente código JSON en un archivo denominado stop_request.json.

   
   {
      "ApplicationName": "test"
   }
   

2. Ejecute la acción StopApplication con la siguiente solicitud para detener la aplicación:

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

Cómo actualizar las propiedades de entorno de la aplicación

1. Guarde el siguiente código JSON en un archivo denominado update_properties_request.json.

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. Ejecute la acción UpdateApplication con la solicitud anterior para actualizar las propiedades del entorno:

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. Elija Configurar.

4. En la sección Instantáneas, seleccione Deshabilitar y, a continuación, seleccione Actualizar.

5. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminació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. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

4. En la página Flujos de Kinesis, elija ExampleOutputStream, elija Acciones, elija Eliminar y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

Procedimiento para registrarse en Cuenta de AWS

1. Abra https://portal.aws.amazon.com/billing/signup.

2. Siga las instrucciones que se le indiquen.

   Parte del procedimiento de registro consiste en recibir una llamada telefónica o mensaje de texto e indicar un código de verificación en el teclado del teléfono.

   Al registrarse en una Cuenta de AWS, se crea un Usuario raíz de la cuenta de AWS. El usuario raíz tendrá acceso a todos los Servicios de AWS y recursos de esa cuenta. Como práctica recomendada de seguridad, asigne acceso administrativo a un usuario y utilice únicamente el usuario raíz para realizar tareas que requieren acceso de usuario raíz.

Protección de Usuario raíz de la cuenta de AWS

1. Inicie sesión en Consola de administración de AWS como propietario de la cuenta; para ello, elija Usuario raíz e introduzca el correo electrónico de su Cuenta de AWS. En la siguiente página, escriba su contraseña.

   Para obtener ayuda para iniciar sesión con el usuario raíz, consulte Iniciar sesión como usuario raíz en la Guía del usuario de AWS Sign-In.

2. Active la autenticación multifactor (MFA) para el usuario raíz.

   Para obtener instrucciones, consulte Habilitación de un dispositivo MFA virtual para su usuario raíz de la Cuenta de AWS (consola) en la Guía del usuario de IAM.

Creación de un usuario con acceso administrativo

1. Activar IAM Identity Center.

   Consulte las instrucciones en Activar AWS IAM Identity Center en la Guía del usuario de AWS IAM Identity Center.

2. En IAM Identity Center, conceda acceso administrativo a un usuario.

   Para ver un tutorial sobre cómo usar Directorio de IAM Identity Center como origen de identidad, consulte Configuración del acceso de los usuarios con el Directorio de IAM Identity Center predeterminado en la Guía del usuario de AWS IAM Identity Center.

Inicio de sesión como usuario con acceso de administrador

• Para iniciar sesión con el usuario de IAM Identity Center, use la URL de inicio de sesión que se envió a la dirección de correo electrónico cuando creó el usuario de IAM Identity Center.

  Para obtener ayuda para iniciar sesión con un usuario de IAM Identity Center, consulte Inicio de sesión en el portal de acceso de AWS en la Guía del usuario de AWS Sign-In.

Concesión de acceso a usuarios adicionales

1. En IAM Identity Center, cree un conjunto de permisos que siga la práctica recomendada de aplicar permisos de privilegios mínimos.

   Para conocer las instrucciones, consulte Create a permission set en la Guía del usuario de AWS IAM Identity Center.

2. Asigne usuarios a un grupo y, a continuación, asigne el acceso de inicio de sesión único al grupo.

   Para conocer las instrucciones, consulte Add groups en la Guía del usuario de AWS IAM Identity Center.

Para configurar la AWS CLI

1. Descargue y configure la AWS CLI. Para obtener instrucciones, consulte los siguientes temas en la Guía del usuario de la AWS Command Line Interface:

   • Instalación del AWS Command Line Interface

   • Configuración de la AWS CLI

2. Añada un perfil con nombre para el usuario administrador en el archivo config de la AWS CLI. Puede utilizar este perfil cuando ejecute los comandos de la AWS CLI. Para obtener más información sobre los perfiles con nombre, consulte Perfiles con nombre en la Guía del usuario de AWS Command Line Interface.

   [profile adminuser]
   aws_access_key_id = adminuser access key ID
   aws_secret_access_key = adminuser secret access key
   region = aws-region

   Para ver una lista de las regiones de AWS disponibles, consulte Regions and Endpoints en la Referencia general de Amazon Web Services.

   nota

   El código y los comandos de ejemplo de este tutorial utilizan la región Oeste de EE. UU. (Oregón). Para usar una región diferente, cambie la región en el código y los comandos de este tutorial por la región que desea usar.

3. Verifique la configuración introduciendo el siguiente comando de ayuda en el símbolo del sistema:

   aws help

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-west-2 \
   --profile adminuser

2. 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-west-2 \
   --profile adminuser

1. Cree un archivo denominado stock.py con el siguiente contenido:

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. Más adelante en el tutorial, se ejecuta el script stock.py para enviar datos a la aplicación.

   $ python stock.py

1. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. Vaya al directorio amazon-kinesis-data-analytics-java-examples/GettingStarted_1_6.

Cómo compilar el código de la aplicación

1. Para utilizar el código de la aplicación, compile y empaquete el código en un archivo JAR. Puede compilar y empaquetar el código de una de las dos formas siguientes:

   • Utilice la herramienta de línea de comandos de Maven. Cree su archivo JAR ejecutando el siguiente comando en el directorio que contiene el archivo pom.xml:

     mvn package

     nota

     El parámetro -DFLink.version no es necesario para la versión 1.0.1 de tiempo de ejecución de Managed Service para Apache Flink; solo es necesario para la versión 1.1.0 y versiones posteriores. Para obtener más información, consulte Especificación de la versión de Apache Flink de su aplicación.

   • Use el entorno de desarrollo. Consulte la documentación de su entorno de desarrollo para obtener más información.

   Puede cargar el paquete como un archivo JAR o puede comprimir el paquete y cargarlo como un archivo ZIP. Si crea su aplicación mediante la AWS CLI, debe especificar el tipo de contenido del código (JAR o ZIP).

2. Si hay errores al compilar, verifique que la variable de entorno JAVA_HOME se ha configurado correctamente.

Cómo cargar el código de la aplicación

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

2. Elija Crear bucket.

3. Escriba ka-app-code-<username> en el campo Nombre del bucket. Añada un sufijo al nombre del bucket, como su nombre de usuario, para que sea único a nivel global. Elija Siguiente.

4. En el paso Configurar opciones, deje los ajustes tal y como están y elija Siguiente.

5. En el paso Establecer permisos, deje los ajustes tal y como están y elija Siguiente.

6. Elija Crear bucket.

7. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y seleccione Subir.

8. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior. Elija Siguiente.

9. En el paso Establecer permisos, deje los ajustes tal y como están. Elija Siguiente.

10. En el paso Establecer propiedades, deje los ajustes tal y como están. Seleccione Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Descripción, escriba My java test app.

   • En Tiempo de ejecución, escriba Apache Flink.

     nota

     Managed Service para Apache Flink utiliza la versión 1.8.2 o 1.6.2 de Apache Flink.

   • Cambie el menú desplegable de versiones a Apache Flink 1.6.

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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:::ka-app-code-username/java-getting-started-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca java-getting-started-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. Escriba las siguientes propiedades y valores de la aplicación:

   ID de grupo	Clave	Valor
   ProducerConfigProperties	flink.inputstream.initpos	LATEST
   ProducerConfigProperties	aws.region	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

5. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

6. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

7. Elija Actualizar.

1. En la página Mi aplicación, elija Ejecutar. Confirme la acción.

2. Cuando la aplicación se está ejecutando, actualice la página. La consola muestra el Gráfico de la aplicación.

Cómo crear un rol de IAM

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

2. En el panel de navegación, elija Roles, Crear rol.

3. En Seleccionar tipo de entidad de confianza, elija Servicio de AWS. En Elegir el servicio que usará este rol, elija Kinesis. En Seleccionar su caso de uso, elija Kinesis Analytics.

   Elija Siguiente: permisos.

4. En la página Asociar políticas de permisos, elija Siguiente: Revisión. Asociará políticas de permisos después de crear el rol.

5. En la página Crear rol, escriba MF-stream-rw-role como Nombre de rol. Elija Crear rol.

   Ahora ha creado un nuevo rol de IAM llamado MF-stream-rw-role. A continuación, actualice las políticas de confianza y permisos del rol.

6. Asocie la política de permisos al rol.

   nota

   Para este ejercicio, Managed Service para Apache Flink asume este rol tanto para leer datos de un flujo de datos de Kinesis (origen) como para escribir la salida en otro flujo de datos de Kinesis. Asocie la política que ha creado en el paso anterior, Creación de una política de permisos.

   1. En la página Resumen, elija la pestaña Permisos.

   2. Seleccione Asociar políticas.

   3. En el campo de búsqueda, escriba AKReadSourceStreamWriteSinkStream (la política que ha creado en la sección anterior).

   4. Elija la política AKReadSourceStreamWriteSinkStream y luego elija Asociar política.

1. Guarde el siguiente código JSON en un archivo denominado create_request.json. Cambie el ARN del rol de ejemplo por el ARN del rol que ha creado antes. Reemplace el sufijo del ARN del bucket (username) por el sufijo que eligió en la sección anterior. Reemplace el ID de la cuenta de ejemplo (012345678901) del rol de ejecución del servicio por el ID de su cuenta.

   {
       "ApplicationName": "test",
       "ApplicationDescription": "my java test app",
       "RuntimeEnvironment": "FLINK-1_6",
       "ServiceExecutionRole": "arn:aws:iam::012345678901:role/MF-stream-rw-role",
       "ApplicationConfiguration": {
           "ApplicationCodeConfiguration": {
               "CodeContent": {
                   "S3ContentLocation": {
                       "BucketARN": "arn:aws:s3:::ka-app-code-username",
                       "FileKey": "java-getting-started-1.0.jar"
                   }
               },
               "CodeContentType": "ZIPFILE"
           },
           "EnvironmentProperties":  { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
       }
   }
   
   

2. Ejecute la acción CreateApplication con la solicitud anterior para crear la aplicación:

   aws kinesisanalyticsv2 create-application --cli-input-json file://create_request.json

Cómo iniciar la aplicación

1. Guarde el siguiente código JSON en un archivo denominado start_request.json.

   {
       "ApplicationName": "test",
       "RunConfiguration": {
           "ApplicationRestoreConfiguration": { 
            "ApplicationRestoreType": "RESTORE_FROM_LATEST_SNAPSHOT"
            }
       }
   }
   

2. Ejecute la acción StartApplication con la solicitud anterior para iniciar la aplicación:

   aws kinesisanalyticsv2 start-application --cli-input-json file://start_request.json

Cómo detener la aplicación

1. Guarde el siguiente código JSON en un archivo denominado stop_request.json.

   {
      "ApplicationName": "test"
   }
   

2. Ejecute la acción StopApplication con la siguiente solicitud para detener la aplicación:

   aws kinesisanalyticsv2 stop-application --cli-input-json file://stop_request.json

Cómo actualizar las propiedades de entorno de la aplicación

1. Guarde el siguiente código JSON en un archivo denominado update_properties_request.json.

   {"ApplicationName": "test",
      "CurrentApplicationVersionId": 1,
      "ApplicationConfigurationUpdate": { 
         "EnvironmentPropertyUpdates": { 
            "PropertyGroups": [ 
               { 
                  "PropertyGroupId": "ProducerConfigProperties",
                  "PropertyMap" : {
                       "flink.stream.initpos" : "LATEST",
                       "aws.region" : "us-west-2",
                       "AggregationEnabled" : "false"
                  }
               },
               { 
                  "PropertyGroupId": "ConsumerConfigProperties",
                  "PropertyMap" : {
                       "aws.region" : "us-west-2"
                  }
               }
            ]
         }
      }
   }
   

2. Ejecute la acción UpdateApplication con la solicitud anterior para actualizar las propiedades del entorno:

   aws kinesisanalyticsv2 update-application --cli-input-json file://update_properties_request.json

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

2. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. Elija Configurar.

4. En la sección Instantáneas, seleccione Deshabilitar y, a continuación, seleccione Actualizar.

5. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminació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. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

4. En la página Flujos de Kinesis, elija ExampleOutputStream, elija Acciones, elija Eliminar y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

1. Cree un archivo denominado stock.py con el siguiente contenido:

   
       import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. Ejecute el script stock.py:

   $ python stock.py

   Mantenga el script en ejecución mientras completa el resto del tutorial.

1. Si aún no lo ha hecho, instale el cliente Git. Para obtener más información, consulte Installing Git.

2. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. Vaya al directorio amazon-kinesis-data-analytics-java-examples/TumblingWindow.

1. Si aún no lo ha hecho, instale Java y Maven. Para obtener más información, consulte Cumplimiento de los requisitos previos obligatorios en el tutorial de Tutorial: introducción al uso de la API de DataStream en Managed Service para Apache Flink.

2. Compile la aplicación con el siguiente comando:

   mvn package -Dflink.version=1.15.3

   nota

   El código fuente proporcionado se basa en bibliotecas de Java 11.

1. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y seleccione Subir.

2. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior.

3. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Tiempo de ejecución, escriba Apache Flink.

     nota

     Managed Service para Apache Flink utiliza la versión 1.15.2 de Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.15.2 (versión recomendada).

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": "logs:DescribeLogStreams",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": "logs:PutLogEvents",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca aws-kinesis-analytics-java-apps-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

5. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

6. Elija Actualizar.

1. En la página Mi aplicación, elija Ejecutar. Deje seleccionada la opción Ejecutar sin instantánea y confirme la acción.

2. Cuando la aplicación se está ejecutando, actualice la página. La consola muestra el Gráfico de 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. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminación.

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

2. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

4. En la página Flujos de Kinesis, elija ExampleOutputStream, elija Acciones, elija Eliminar y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

1. Cree un archivo denominado stock.py con el siguiente contenido:

    
   import datetime
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_data():
       return {
           "EVENT_TIME": datetime.datetime.now().isoformat(),
           "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
           "PRICE": round(random.random() * 100, 2),
       }
   
   
   def generate(stream_name, kinesis_client):
       while True:
           data = get_data()
           print(data)
           kinesis_client.put_record(
               StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

2. Ejecute el script stock.py:

   $ python stock.py

   Mantenga el script en ejecución mientras completa el resto del tutorial.

1. Si aún no lo ha hecho, instale el cliente Git. Para obtener más información, consulte Installing Git.

2. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. Vaya al directorio amazon-kinesis-data-analytics-java-examples/SlidingWindow.

1. Si aún no lo ha hecho, instale Java y Maven. Para obtener más información, consulte Cumplimiento de los requisitos previos obligatorios en el tutorial de Tutorial: introducción al uso de la API de DataStream en Managed Service para Apache Flink.

2. Compile la aplicación con el siguiente comando:

   mvn package -Dflink.version=1.15.3

   nota

   El código fuente proporcionado se basa en bibliotecas de Java 11.

1. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y luego seleccione Cargar.

2. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior.

3. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Tiempo de ejecución, escriba Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.15.2 (versión recomendada).

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": "logs:DescribeLogStreams",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": "logs:PutLogEvents",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca aws-kinesis-analytics-java-apps-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

5. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

6. Elija Actualizar.

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. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminación.

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

2. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

4. En la página Flujos de Kinesis, elija ExampleOutputStream, elija Acciones, elija Eliminar y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

1. Cree un archivo denominado stock.py con el siguiente contenido:

   
       import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. Ejecute el script stock.py:

   $ python stock.py

   Mantenga el script en ejecución mientras completa el resto del tutorial.

1. Si aún no lo ha hecho, instale el cliente Git. Para obtener más información, consulte Installing Git.

2. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. Vaya al directorio amazon-kinesis-data-analytics-java-examples/S3Sink.

1. Si aún no lo ha hecho, instale Java y Maven. Para obtener más información, consulte Cumplimiento de los requisitos previos obligatorios en el tutorial de Tutorial: introducción al uso de la API de DataStream en Managed Service para Apache Flink.

2. Compile la aplicación con el siguiente comando:

   mvn package -Dflink.version=1.15.3

1. En la consola de Amazon S3, elija el bucket ka-app-code-<username>, navegue a la carpeta código y seleccione Cargar.

2. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior.

3. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Tiempo de ejecución, escriba Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.15.2 (versión recomendada).

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

   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:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Tiempo de ejecución, escriba Apache Flink.

   • Deje la versión como Apache Flink, versión 1.15.2 (versión recomendada).

6. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

7. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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. Sustituya <username> por el nombre de usuario.

   {
               "Sid": "S3",
               "Effect": "Allow",
               "Action": [
                   "s3:Abort*",
                   "s3:DeleteObject*",
                   "s3:GetObject*",
                   "s3:GetBucket*",
                   "s3:List*",
                   "s3:ListBucket",
                   "s3:PutObject"
               ],
               "Resource": [
                   "arn:aws:s3:::ka-app-code-<username>",
                   "arn:aws:s3:::ka-app-code-<username>/*"
               ]
             }, 
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:*"
               ]
           },
           {
               "Sid": "ListCloudwatchLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:%LOG_GROUP_PLACEHOLDER%:log-stream:*"
               ]
           },
           {
               "Sid": "PutCloudwatchLogs",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:region:account-id:log-group:%LOG_GROUP_PLACEHOLDER%:log-stream:%LOG_STREAM_PLACEHOLDER%"
               ]
           }
           ,
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           
       ]
   }

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca code/aws-kinesis-analytics-java-apps-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

5. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

6. Elija Actualizar.

1. En la página Mi aplicación, elija Ejecutar. Deje seleccionada la opción Ejecutar sin instantánea y confirme la acción.

2. Cuando la aplicación se está ejecutando, actualice la página. La consola muestra el Gráfico de la aplicación.

1. Añada las siguientes instrucciones de importación al principio del archivo S3StreamingSinkJob.java:

   import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
   import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;

2. Actualice el método createS3SinkFromStaticConfig() en el código para que tenga el siguiente aspecto:

   private static StreamingFileSink<String> createS3SinkFromStaticConfig() {
   
       final StreamingFileSink<String> sink = StreamingFileSink
           .forRowFormat(new Path(s3SinkPath), new SimpleStringEncoder<String>("UTF-8"))
           .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd--HH"))
           .withRollingPolicy(DefaultRollingPolicy.create().build())
           .build();
       return sink;
   }
               

1. Aumente el parámetro CheckpointInterval de la aplicación. La siguiente entrada para la acción UpdateApplication establece el intervalo del punto de control en 10 minutos:

   {
      "ApplicationConfigurationUpdate": {  
         "FlinkApplicationConfigurationUpdate": { 
            "CheckpointConfigurationUpdate": {
               "ConfigurationTypeUpdate" : "CUSTOM", 
               "CheckpointIntervalUpdate": 600000
            }
         }
      },
      "ApplicationName": "MyApplication",
      "CurrentApplicationVersionId": 5
   }

   Para utilizar el código anterior, especifique la versión actual de la aplicación. Se puede recuperar la versión de la aplicación mediante la acción ListApplications.

2. Añada la siguiente instrucción de importación al principio del archivo S3StreamingSinkJob.java:

   import java.util.concurrent.TimeUnit; 

3. Actualice el método createS3SinkFromStaticConfig en el archivo S3StreamingSinkJob.java para que tenga el siguiente aspecto:

   private static StreamingFileSink<String> createS3SinkFromStaticConfig() {
   
           final StreamingFileSink<String> sink = StreamingFileSink
                   .forRowFormat(new Path(s3SinkPath), new SimpleStringEncoder<String>("UTF-8"))
                   .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd--HH"))
                   .withRollingPolicy(
                           DefaultRollingPolicy.create()
                               .withRolloverInterval(TimeUnit.MINUTES.toMillis(8))
                               .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
                               .withMaxPartSize(1024 * 1024 * 1024)
                               .build())
                   .build();
           return sink;
       }

   El ejemplo de código anterior establece la frecuencia de escrituras en el bucket de Amazon S3 en 8 minutos.

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. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminación.

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

2. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

1. Si aún no lo ha hecho, instale el cliente Git. Para obtener más información, consulte Installing Git.

2. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. El código de la aplicación se encuentra en el archivo amazon-kinesis-data-analytics-java-examples/KafkaConnectors/KafkaGettingStartedJob.java. Puede examinar el código para familiarizarse con la estructura del código de la aplicación Managed Service para Apache Flink.

4. Utilice la herramienta Maven de línea de comandos o el entorno de desarrollo que prefiera para crear el archivo JAR. Para compilar el archivo JAR utilizando la herramienta Maven de línea de comandos, introduzca lo siguiente:

   mvn package -Dflink.version=1.15.3

   Si la compilación es correcta, se crea el siguiente archivo:

   target/KafkaGettingStartedJob-1.0.jar

   nota

   El código fuente proporcionado se basa en bibliotecas de Java 11. Si está utilizando un entorno de desarrollo,

1. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y seleccione Subir.

2. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo KafkaGettingStartedJob-1.0.jar que creó en el paso anterior.

3. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Tiempo de ejecución, escriba Apache Flink versión 1.15.2.

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca KafkaGettingStartedJob-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

   nota

   Al especificar los recursos de la aplicación mediante la consola (como CloudWatch Logs o una Amazon VPC), la consola modifica el rol de ejecución de la aplicación para conceder permiso de acceso a esos recursos.

4. En Propiedades, elija Añadir grupo. Introduzca las siguientes propiedades:

   ID de grupo	Clave	Valor
   KafkaSource	tema	AWSKafkaTutorialTopic
   KafkaSource	bootstrap.servers	La lista de servidores de arranque que guardó anteriormente
   KafkaSource	security.protocol	SSL
   KafkaSource	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSource	ssl.truststore.contraseña	changeit

   nota

   El ssl.truststore.password del certificado predeterminado es “changeit”; no necesita cambiar este valor si utiliza el certificado predeterminado.

   Vuelva a seleccionar Añadir grupo. Introduzca las siguientes propiedades:

   ID de grupo	Clave	Valor
   KafkaSink	tema	AWSKafkaTutorialTopicDestination
   KafkaSink	bootstrap.servers	La lista de servidores de arranque que guardó anteriormente
   KafkaSink	security.protocol	SSL
   KafkaSink	ssl.truststore.location	/usr/lib/jvm/java-11-amazon-corretto/lib/security/cacerts
   KafkaSink	ssl.truststore.contraseña	changeit
   KafkaSink	transacción.timeout.ms	1 000

   El código de la aplicación lee las propiedades de la aplicación anteriores para configurar el origen y el receptor que se utilizan para interactuar con la VPC y el clúster de Amazon MSK. Para obtener más información acerca de las propiedades, consulte Uso de propiedades de tiempo de ejecución.

5. En Instantáneas, seleccione Desactivar. Esto facilitará la actualización de la aplicación sin cargar datos de estado de la aplicación no válidos.

6. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

7. En Registro de CloudWatch, elija la casilla de verificación Habilitar.

8. En la sección Nube privada virtual (VPC), elija la VPC que desea asociar a la aplicación. Elija las subredes y el grupo de seguridad asociados a la VPC que desee que utilice la aplicación para acceder a los recursos de la VPC.

9. Elija Actualizar.

1. Cree un archivo denominado stock.py con el siguiente contenido:

    
       import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. Ejecute el script stock.py:

   $ python stock.py

   Mantenga el script en ejecución mientras completa el resto del tutorial.

1. Si aún no lo ha hecho, instale el cliente Git. Para obtener más información, consulte Installing Git.

2. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

3. Vaya al directorio amazon-kinesis-data-analytics-java-examples/EfoConsumer.

1. Si aún no lo ha hecho, instale Java y Maven. Para obtener más información, consulte Cumplimiento de los requisitos previos obligatorios en el tutorial de Tutorial: introducción al uso de la API de DataStream en Managed Service para Apache Flink.

2. Compile la aplicación con el siguiente comando:

   mvn package -Dflink.version=1.15.3

   nota

   El código fuente proporcionado se basa en bibliotecas de Java 11.

1. En la consola de Amazon S3, elija el bucket ka-app-code-<username> y seleccione Subir.

2. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo aws-kinesis-analytics-java-apps-1.0.jar que creó en el paso anterior.

3. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Tiempo de ejecución, escriba Apache Flink.

     nota

     Managed Service para Apache Flink utiliza la versión 1.15.2 de Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.15.2 (versión recomendada).

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione 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.

   nota

   Estos permisos otorgan a la aplicación la posibilidad de acceder al consumidor de EFO.

   JSON

   {
       "Version":"2012-10-17",		 	 	 
       "Statement": [
           {
               "Sid": "ReadCode",
               "Effect": "Allow",
               "Action": [
                   "s3:GetObject",
                   "logs:DescribeLogGroups",
                   "s3:GetObjectVersion"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*",
                   "arn:aws:s3:::ka-app-code-<username>/aws-kinesis-analytics-java-apps-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": "logs:DescribeLogStreams",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": "logs:PutLogEvents",
               "Resource": "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
           },
           {
               "Sid": "ListCloudwatchLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "AllStreams",
               "Effect": "Allow",
               "Action": [
                   "kinesis:ListShards",
                   "kinesis:ListStreamConsumers",
                   "kinesis:DescribeStreamSummary"
               ],
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/*"
           },
           {
               "Sid": "Stream",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStream",
                   "kinesis:RegisterStreamConsumer",
                   "kinesis:DeregisterStreamConsumer"
               ],
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteOutputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleOutputStream"
           },
           {
               "Sid": "Consumer",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:SubscribeToShard"
               ],
               "Resource": [
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app",
                   "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream/consumer/my-efo-flink-app:*"
               ]
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca aws-kinesis-analytics-java-apps-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. En Propiedades, elija Crear grupo.

5. Escriba las siguientes propiedades y valores de la aplicación:

   ID de grupo	Clave	Valor
   ConsumerConfigProperties	flink.stream.recordpublisher	EFO
   ConsumerConfigProperties	flink.stream.efo.consumername	basic-efo-flink-app
   ConsumerConfigProperties	INPUT_STREAM	ExampleInputStream
   ConsumerConfigProperties	flink.inputstream.initpos	LATEST
   ConsumerConfigProperties	AWS_REGION	us-west-2

6. En Propiedades, elija Crear grupo.

7. Escriba las siguientes propiedades y valores de la aplicación:

   ID de grupo	Clave	Valor
   ProducerConfigProperties	OUTPUT_STREAM	ExampleOutputStream
   ProducerConfigProperties	AWS_REGION	us-west-2
   ProducerConfigProperties	AggregationEnabled	false

8. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

9. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

10. Elija Actualizar.

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. En el panel de Managed Service para Apache Flink, elija MyApplication.

3. En la página de la aplicación, seleccione Eliminar y, a continuación, confirme la eliminación.

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

2. En el panel de Kinesis Data Streams, elija ExampleInputStream.

3. En la página ExampleInputStream, elija Eliminar flujo de Kinesis y, a continuación, confirme la eliminación.

4. En la página Flujos de Kinesis, elija ExampleOutputStream, elija Acciones, elija Eliminar y, a continuación, confirme la eliminación.

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

2. Elija el bucket ka-app-code-<username>.

3. Elija Eliminar y luego ingrese el nombre del bucket para confirmar la eliminación.

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

2. En la barra de navegación, seleccione Políticas.

3. En el control de filtros, introduzca kinesis.

4. Elija la política kinesis-analytics-service-MyApplication-us-west-2.

5. Seleccione Acciones de política y, a continuación, Eliminar.

6. En la barra de navegación, seleccione Roles.

7. Elija el rol kinesis-analytics-MyApplication-us-west-2.

8. Elija Eliminar rol y, a continuación, confirme la eliminación.

1. Abra la consola de CloudWatch en https://console.aws.amazon.com/cloudwatch/.

2. En la barra de navegación, elija Registros.

3. Elija el grupo de registro /aws/kinesis-analytics/MyApplication.

4. Elija Eliminar grupo de registro y, a continuación, confirme la eliminación.

1. Cree un archivo denominado stock.py con el siguiente contenido:

    import datetime
       import json
       import random
       import boto3
   
       STREAM_NAME = "ExampleInputStream"
   
   
       def get_data():
           return {
               'event_time': datetime.datetime.now().isoformat(),
               'ticker': random.choice(['AAPL', 'AMZN', 'MSFT', 'INTC', 'TBV']),
               'price': round(random.random() * 100, 2)}
   
   
       def generate(stream_name, kinesis_client):
           while True:
               data = get_data()
               print(data)
               kinesis_client.put_record(
                   StreamName=stream_name,
                   Data=json.dumps(data),
                   PartitionKey="partitionkey")
   
   
       if __name__ == '__main__':
           generate(STREAM_NAME, boto3.client('kinesis', region_name='us-west-2'))

2. Ejecute el script stock.py:

   $ python stock.py

   Mantenga el script en ejecución mientras completa el resto del tutorial.

1. Clone el repositorio remoto con el siguiente comando:

   git clone https://github.com/aws-samples/amazon-kinesis-data-analytics-examples.git

2. Vaya al directorio amazon-kinesis-data-analytics-java-examples/FirehoseSink.

1. Si aún no lo ha hecho, instale Java y Maven. Para obtener más información, consulte Cumplimiento de los requisitos previos obligatorios en el tutorial de Tutorial: introducción al uso de la API de DataStream en Managed Service para Apache Flink.

2. Para utilizar el conector de Kinesis para la siguiente aplicación, debe descargar, compilar e instalar Apache Maven. Para obtener más información, consulte Uso del conector Apache Flink Kinesis Streams con versiones anteriores de Apache Flink.

3. Compile la aplicación con el siguiente comando:

   mvn package -Dflink.version=1.15.3

   nota

   El código fuente proporcionado se basa en bibliotecas de Java 11.

Cómo cargar el código de la aplicación

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

2. En la consola, elija el bucket ka-app-code-<username> y elija Cargar.

3. En el paso Seleccionar archivos, elija Añadir archivos. Vaya al archivo java-getting-started-1.0.jar que creó en el paso anterior.

4. No es necesario cambiar ninguno de los ajustes del objeto, por lo tanto, elija Cargar.

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. En el panel de Managed Service para Apache Flink, seleccione Crear aplicación de análisis.

3. En la página Managed Service para Apache Flink: crear aplicación, proporcione los detalles de la aplicación de la siguiente manera:

   • En Nombre de la aplicación, escriba MyApplication.

   • En Descripción, escriba My java test app.

   • En Tiempo de ejecución, escriba Apache Flink.

     nota

     Managed Service para Apache Flink utiliza la versión 1.15.2 de Apache Flink.

   • Deje el menú desplegable de versión como Apache Flink versión 1.15.2 (versión recomendada).

4. Para los permisos de acceso, seleccione Crear o actualizar el rol de IAM. kinesis-analytics-MyApplication-us-west-2

5. Elija Crear aplicación.

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-west-2 que la consola creó en su nombre en la sección anterior.

3. En la página Resumen, elija Editar política. Seleccione la pestaña JSON.

4. Añada la sección subrayada de la siguiente política de ejemplo a la política. Reemplace todas las instancias del 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:::ka-app-code-username/java-getting-started-1.0.jar"
               ]
           },
           {
               "Sid": "DescribeLogGroups",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogGroups"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:*"
               ]
           },
           {
               "Sid": "DescribeLogStreams",
               "Effect": "Allow",
               "Action": [
                   "logs:DescribeLogStreams"
              ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:*"
               ]
           },
           {
               "Sid": "PutLogEvents",
               "Effect": "Allow",
               "Action": [
                   "logs:PutLogEvents"
               ],
               "Resource": [
                   "arn:aws:logs:us-west-2:012345678901:log-group:/aws/kinesis-analytics/MyApplication:log-stream:kinesis-analytics-log-stream"
               ]
           },
           {
               "Sid": "ReadInputStream",
               "Effect": "Allow",
               "Action": "kinesis:*",
               "Resource": "arn:aws:kinesis:us-west-2:012345678901:stream/ExampleInputStream"
           },
           {
               "Sid": "WriteDeliveryStream",
               "Effect": "Allow",
               "Action": "firehose:*",
               "Resource": "arn:aws:firehose:us-west-2:012345678901:deliverystream/ExampleDeliveryStream"
           }
       ]
   }
   

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

2. En la página Configurar aplicación, proporcione la Ubicación del código:

   • Para el bucket de Amazon S3, introduzca ka-app-code-<username>.

   • En Ruta al objeto de Amazon S3, introduzca java-getting-started-1.0.jar.

3. En Acceso a los recursos de la aplicación, en Permisos de acceso, seleccione Crear o actualizar el rol de IAM kinesis-analytics-MyApplication-us-west-2.

4. En Monitorización, asegúrese de que el Nivel de métricas de monitorización se ha establecido en Aplicación.

5. En Registro de CloudWatch, seleccione la casilla de verificación Habilitar.

6. Elija Actualizar.

Cómo crear un rol de IAM

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

2. En el panel de navegación, elija Roles, Crear rol.

3. En Seleccionar tipo de entidad de confianza, elija Servicio de AWS. En Elegir el servicio que usará este rol, elija Kinesis. En Seleccionar su caso de uso, elija Kinesis Analytics.

   Elija Siguiente: permisos.

4. En la página Asociar políticas de permisos, elija Siguiente: Revisión. Asociará políticas de permisos después de crear el rol.

5. En la página Crear rol, escriba MF-stream-rw-role como Nombre de rol. Elija Crear rol.

   Ahora ha creado un nuevo rol de IAM llamado MF-stream-rw-role. A continuación, actualice las políticas de confianza y permisos del rol.

6. Asocie la política de permisos al rol.

   nota

   Para este ejercicio, Managed Service para Apache Flink asume este rol tanto para leer datos de un flujo de datos de Kinesis (origen) como para escribir la salida en otro flujo de datos de Kinesis. Asocie la política que ha creado en el paso anterior, Creación de una política de permisos.

   1. En la página Resumen, elija la pestaña Permisos.

   2. Seleccione Asociar políticas.

   3. En el campo de búsqueda, escriba AKReadSourceStreamWriteSinkStream (la política que ha creado en la sección anterior).

   4. Elija la política AKReadSourceStreamWriteSinkStream y luego elija Asociar política.