Kinesis Data AnalyticsをKinesis Data Streamsに接続してSQL検索する

2021/08/11 2021/08/11

Amazon Kinesis Data StreamsにTwitter検索データを送信するで収集しているツイートをKinesis Data AnalyticsでSQL検索してみます。

Kinesis Data Analyticsアプリケーションの作成

アプリケーションの名前を決めて検索方法をSQLにしました。

Kinesis Data Streamsに接続

[ストリーミングデータを接続]を押下しました。

作成済のKinesis Data Streamを選択しました。

スキーマを検出を実行しました。

検出できました。

SQLを実行

[Go to SQL editor]ボタンを押下しました。

アプリケーション開始のメッセージが表示されるので、[Yes, start application]を押下しました。

テンプレートにサンプルコードがあったのでこれを少し変えました。

-- Create output stream, which can be used to send to a destination
CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" 
("text" VARCHAR(256),"source" VARCHAR(256), "user_screen_name" VARCHAR(256),"user_name" VARCHAR(256), "user_followers_count" INTEGER);
-- Create pump to insert into output 
CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
-- Select all columns from source stream
SELECT STREAM "text", "source", "user_screen_name", "user_name", "user_followers_count"
FROM "SOURCE_SQL_STREAM_001"
-- LIKE compares a string to a string pattern (_ matches all char, % matches substring)
-- SIMILAR TO compares string to a regex, may use ESCAPE
WHERE "text" SIMILAR TO '%合格%';

-- Create output stream, which can be used to send to a destination

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM"

("text" VARCHAR(256),"source" VARCHAR(256), "user_screen_name" VARCHAR(256),"user_name" VARCHAR(256), "user_followers_count" INTEGER);

-- Create pump to insert into output

CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"

-- Select all columns from source stream

SELECT STREAM "text", "source", "user_screen_name", "user_name", "user_followers_count"

FROM "SOURCE_SQL_STREAM_001"

-- LIKE compares a string to a string pattern (_ matches all char, % matches substring)

-- SIMILAR TO compares string to a regex, may use ESCAPE

WHERE "text" SIMILAR TO '%合格%';

そして、再度Twitterからデータを取得してKinesis Data Streamに流してみました。
合格の嬉しい声がたくさん検索されてDESTINATION_SQL_STREAMにINSERTされました。

Twitterの高度検索が思い通りにいかないなあという場合は、Twitterをシンプルに検索した結果をKinesisに流して、Analyticsで検索するのもありですね。

次回はこの結果をKinesis Data FirehoseからS3バケットに保存します。

最後までお読みいただきましてありがとうございました！

「AWS認定資格試験テキスト＆問題集　AWS認定ソリューションアーキテクト - プロフェッショナル改訂第2版」という本を書きました。

「AWS認定資格試験テキスト AWS認定クラウドプラクティショナー改訂第3版」という本を書きました。

「AWS認定資格試験テキスト　AWS認定AIプラクティショナー」という本を書きました。

「ポケットスタディ AWS認定デベロッパーアソシエイト［DVA-C02対応］」という本を書きました。

「要点整理から攻略するAWS認定ソリューションアーキテクト-アソシエイト」という本を書きました。

「AWSではじめるLinux入門ガイド」という本を書きました。

@yamamanx

開発ベンダー5年、ユーザ企業システム部門通算9年、ITインストラクター5年目でプロトタイプビルダーもやりだしたSoftware Engineerです。
質問はコメントかSNSなどからお気軽にどうぞ。
出来る限りなるべく答えます。

このブログの内容/発言の一切は個人の見解であり、所属する組織とは関係ありません。
このブログは経験したことなどの共有を目的としており、手順や結果などを保証するものではありません。
ご参考にされる際は、読者様自身のご判断にてご対応をお願いいたします。

また、勉強会やイベントのレポートは自分が気になったことをメモしたり、聞いて思ったことを書いていますので、登壇者の意見や発表内容ではありません。

- AWS AWS, kinesis

: Cloud9でCodeWhispererを使用する

アクセス権限 Cloud9のEC2に設定するIAMロールのIAMポリシーで co …

: API GatewayをトリガーにしたときのLambdaリソースベースポリシー

先日、SwaggerからAPI Gatewayを作ったときに、API Gatew …

: AWS認定試験の自宅受験で壁のポスターを注意されちゃいました

AWS認定オンライン受験をしてみましたに書きましたとおり、自宅受験デビューしまし …

: AWS Lambdaで「Process exited before completing request」

AWS lambdaで「Process exited before comple …

: ヤマムギ vol.7 AWSアカウント作成 & 最初の設定ハンズオン手順

ヤマムギとは from Mitsuhiro Yamashita 「AWSではじめ …

: サイトのHTTPステータスを5分おきにチェックして200以外ならSlackに通知する

すいません。ここ最近出費が重なりまして、某監視サービスのプランを有料プランからF …

: リージョナルNATゲートウェイを確認しました

2025年11月にアップデートが公開された、リージョナルNATゲートウェイの動作 …

: Amazon EC2 インスタンスメタデータサービス(IMDS) v2のみにしました

今さらながらですが、EC2 インスタンスメタデータサービス(IMDS) v2を確 …

: AWS Lambdaで Unable to import module エラーが発生したときは

そもそも、Pycharmのナビゲーションペインでディレクトリごとドラッグ&amp …

: RocketChat(EC2インスタンス)でCPU80%以上を10分間継続したら再起動する

先日、数日間のやり取り用で完全に使い捨てとして使っているRocketChatで、 …

PREV: PostmanでTwitter API 1.1を使う
NEXT: Kinesis Data Analyticsの検索結果をKinesis Data FirehoseからS3バケットに保存する

Kinesis Data AnalyticsをKinesis Data Streamsに接続してSQL検索する

Kinesis Data Analyticsアプリケーションの作成

Kinesis Data Streamsに接続

SQLを実行

関連記事