ステップ 1: 準備 - Amazon Kinesis Data Analytics for SQL Applications デベロッパーガイド
ステップ 1.1: 入力ストリームと出力データストリームを作成するステップ 1.2: 入力ストリームにサンプルレコードを書き込みます

慎重に検討した結果、アプリケーションの Amazon Kinesis Data Analytics は 2 つのステップSQLで中止することにしました。

1. 2025 年 10 月 15 日以降、SQLアプリケーションの新しい Kinesis Data Analytics を作成することはできません。

2. 2026 年 1 月 27 日以降、アプリケーションを削除します。SQL アプリケーションの Amazon Kinesis Data Analytics を開始または操作することはできません。SQL それ以降、Amazon Kinesis Data Analytics のサポートは終了します。詳細については、「Amazon Kinesis Data Analytics for SQL Applications の中止」を参照してください。

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

ステップ 1: 準備

この実習用の Amazon Kinesis Data Analytics アプリケーションを作成する前に、2 つの Kinesis データストリームを作成する必要があります。ストリームの 1 つはアプリケーションのストリーミングソースとして設定し、もう 1 つのストリームは Kinesis Data Analytics がアプリケーション出力を永続化する宛先として設定します。

ステップ 1.1: 入力ストリームと出力データストリームを作成する

このセクションでは、2 つの Kinesis ストリーム (ExampleInputStream および ExampleOutputStream) を作成します。AWS Management Console または AWS CLI を使用してこれらのストリームを作成できます。

  • コンソールを使用するには

    1. AWS Management Console にサインインし、Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

    2. [データストリームの作成] を選択します。ExampleInputStream という名前の 1 つのシャードがあるストリームを作成します。詳細については、「Amazon Kinesis Data Streams デベロッパーガイド」の「ストリームを作成する」を参照してください。

    3. 前のステップを繰り返し、ExampleOutputStream という名前の 1 つのシャードを持つストリームを作成します。

  • AWS CLI を使用するには

    1. 次の Amazon Kinesis create-stream AWS CLI コマンドを使用して、1 つ目のストリーム (ExampleInputStream) を作成します。

      $ aws kinesis create-stream \
--stream-name ExampleInputStream \
--shard-count 1 \
--region us-east-1 \
--profile adminuser

    2. 同じコマンドを実行し、ストリーム名を ExampleOutputStream に変更します。このコマンドは、アプリケーションが出力の書き込みに使用する 2 つ目のストリームを作成します。

ステップ 1.2: 入力ストリームにサンプルレコードを書き込みます

このステップでは、Python コードを実行してサンプルレコードを連続生成し、それらのレコードを ExampleInputStream ストリームに書き込みます。

{"heartRate": 60, "rateType":"NORMAL"} 
...
{"heartRate": 180, "rateType":"HIGH"}

  1. Python および pip をインストールします。

    Python のインストールについては、Python ウェブサイトをご覧ください。

    pip を使用して依存関係をインストールできます。pip のインストールについては、pip ウェブサイトの「Installation」を参照してください。

  2. 以下の Python コードを実行します。コードの put-record コマンドは、ストリームに JSON レコードを書き込みます。

     
from enum import Enum
import json
import random
import boto3

STREAM_NAME = "ExampleInputStream"


class RateType(Enum):
    normal = "NORMAL"
    high = "HIGH"


def get_heart_rate(rate_type):
    if rate_type == RateType.normal:
        rate = random.randint(60, 100)
    elif rate_type == RateType.high:
        rate = random.randint(150, 200)
    else:
        raise TypeError
    return {"heartRate": rate, "rateType": rate_type.value}


def generate(stream_name, kinesis_client, output=True):
    while True:
        rnd = random.random()
        rate_type = RateType.high if rnd < 0.01 else RateType.normal
        heart_rate = get_heart_rate(rate_type)
        if output:
            print(heart_rate)
        kinesis_client.put_record(
            StreamName=stream_name,
            Data=json.dumps(heart_rate),
            PartitionKey="partitionkey",
        )


if __name__ == "__main__":
    generate(STREAM_NAME, boto3.client("kinesis"))

次のステップ

ステップ 2: アプリケーションの作成