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

: JAWS FESTA 2017 Reverse X re:Birth

JAWS FESTA 2017 中四国今年はブログタイトル「JAWS FEST …

: Active Directory認証でAWSマネジメントコンソールにSSO

こちらの2つのサイトを参考にすすめました。 Active Directory資産 …

: ヤマムギvol.27 Amazon Route 53プライベートホストゾーンとリゾルバーのデモをしました

今日は『AWS認定資格試験テキスト&問題集AWS認定ソリューションアーキ …

: AWS VPC ネットワークACLでHTTP(80)のみ許可する

ユーザーガイドのカスタムネットワーク ACLのうち、HTTP(80)を許可する設 …

: 「Label Training Data for Machine Learning」チュートリアルの記録

Label Training Data for Machine Learning …

: AWS CDK とにかくサンプルでやってみる

Cloud9でAWS CDK環境で作った環境でサンプルプロジェクトからのデプロイ …

: ハンズオン: IAMロール、Secrets Manager、KMS ~AWS認定デベロッパーアソシエイト(DCA-C02)~

このブログは2026年6月29日翔泳社さんより発売される「AWS教科書 AWS認 …

: S3インベントリ設定でインベントリファイルの作成を設定

インベントリレポートファイルはオブジェクトの一覧情報です。日次、週次で定期作成 …

: AWSアカウント内特定リージョン内のDynamoDBテーブルを全削除するLambda(Python)

やりたいことタイトル通りですが、特定アカウントの特定リージョン内のDynamo …

: EC2ユーザーデータからメタデータを取得してRocket.Chatで80ポートを使用する

Rocket.Chatのデフォルトポート番号は3000です。 80を使うようにす …

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を実行

関連記事