Étape 2 : Création de l'application

Dans cette section, vous allez créer une application Kinesis Data Analytics. Vous mettez ensuite à jour l'application en ajoutant une configuration d'entrée qui mappe la source de diffusion que vous avez créée dans la section précédente à un flux d'entrée intégré à l'application.

1. Ouvrez la console du service géré pour Apache Flink à l’adresse https://console.aws.amazon.com/kinesisanalytics.

2. Choisissez Créer une application. Cet exemple utilise le nom de l'application ProcessMultipleRecordTypes.

3. Sur la page de détails de l'application, choisissez Connect streaming data (Connecter des données de diffusion) pour vous connecter à la source.

4. Sur la page Connect to source (Se connecter à la source), procédez comme suit :

   1. Choisissez le flux que vous avez créé dans Étape 1 : Préparation des données.

   2. Choisissez de créer un rôle IAM.

   3. Attendez que la console affiche le schéma déduit et les exemples d'enregistrements qui sont utilisés pour déduire le schéma pour le flux intégré à l'application créé.

   4. Choisissez Save and continue (Enregistrer et continuer).

5. Dans le hub d'applications, choisissez Go to SQL editor. Pour lancer l'application, choisissez Yes, start application (Oui, démarrer l'application) dans la boîte de dialogue qui s'affiche.

6. Dans l'éditeur SQL, écrivez le code de l'application et vérifiez les résultats :

   1. Copiez le code d'application suivant et collez-le dans l'éditeur.

      --Create Order_Stream.
      CREATE OR REPLACE STREAM "Order_Stream" 
                 ( 
                  "order_id"     integer, 
                  "order_type"   varchar(10),
                  "ticker"       varchar(4),
                  "order_price"  DOUBLE, 
                  "record_type"  varchar(10)
                  );
      
      CREATE OR REPLACE PUMP "Order_Pump" AS 
         INSERT INTO "Order_Stream"
            SELECT STREAM "Oid", "Otype","Oticker", "Oprice", "RecordType" 
            FROM   "SOURCE_SQL_STREAM_001"
            WHERE  "RecordType" = 'Order';
      --********************************************
      --Create Trade_Stream.      
      CREATE OR REPLACE STREAM "Trade_Stream" 
                 ("trade_id"     integer, 
                  "order_id"     integer, 
                  "trade_price"  DOUBLE, 
                  "ticker"       varchar(4),
                  "record_type"  varchar(10)
                  );
      
      CREATE OR REPLACE PUMP "Trade_Pump" AS 
         INSERT INTO "Trade_Stream"
            SELECT STREAM "Tid", "Toid", "Tprice", "Tticker", "RecordType"
            FROM   "SOURCE_SQL_STREAM_001"
            WHERE  "RecordType" = 'Trade';
      --*****************************************************************
      --do some analytics on the Trade_Stream and Order_Stream. 
      CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
                  "ticker"  varchar(4),
                  "trade_count"   integer
                  );
      
      CREATE OR REPLACE PUMP "Output_Pump" AS 
         INSERT INTO "DESTINATION_SQL_STREAM"
            SELECT STREAM "ticker", count(*) as trade_count
            FROM   "Trade_Stream"
            GROUP BY "ticker",
                      FLOOR("Trade_Stream".ROWTIME TO MINUTE);

   2. Choisissez Save and run SQL (Enregistrer et exécuter SQL). Choisissez l'onglet Real-time analytics (Analyse en temps réel) pour voir tous les flux intégrés à l'application que l'application a créés et vérifier les données.

Étape suivante

Vous pouvez configurer la sortie de l'application pour conserver les résultats vers une destination externe, telle qu'un autre flux Kinesis ou un flux de diffusion de données Firehose.