Pour les nouveaux projets, nous vous recommandons d’utiliser le nouveau service géré pour Apache Flink Studio plutôt que les applications Kinesis Data Analytics pour SQL. Le service géré pour Apache Flink Studio allie facilité d’utilisation et capacités analytiques avancées, ce qui vous permet de créer des applications sophistiquées de traitement des flux en quelques minutes.
Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
É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.
Ouvrez la console du service géré pour Apache Flink à l’adresse https://console.aws.amazon.com/kinesisanalytics
. -
Choisissez Créer une application. Cet exemple utilise le nom de l'application
ProcessMultipleRecordTypes
. -
Sur la page de détails de l'application, choisissez Connect streaming data (Connecter des données de diffusion) pour vous connecter à la source.
-
Sur la page Connect to source (Se connecter à la source), procédez comme suit :
-
Choisissez le flux que vous avez créé dans Étape 1 : Préparation des données.
-
Choisissez de créer un rôle IAM.
-
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éé.
-
Choisissez Save and continue (Enregistrer et continuer).
-
-
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.
-
Dans l'éditeur SQL, écrivez le code de l'application et vérifiez les résultats :
-
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);
-
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.