Amazon RedshiftからAmazon Kinesisのストリームデータをニアリアルタイムにインジェストしてみた

Amazon RedshiftからAmazon Kinesis Data Streamのストリームデータをマテリアライズド・ビュー経由でニアリアルタイムにインジェストする機能が2022年2月からパブリック・プレビュー提供されています。

Amazon Redshift announces public preview of Streaming Ingestion for Kinesis Data Streams

Amazon Kinesis Data StreamのストリームデータをRedshiftから利用したい場合、従来は Amazon Kinesis Data Firehoseを経由して一度 S3 に出力し、S3 データを Redshift に COPY する必要があり、ストイームデータのリアルタイム性が損なわれていました。

本機能を利用すると、マテリアライズド・ビューの更新レイテンシーが発生するものの、ニアリアルタイムにKinesis Data Streamsのデータをインジェストできます。

やってみた

Kinesis Data StreamsにPUTされたレコードをRedshiftからSQLで参照するところまでを動作確認します。

1. Kinesis Data Stream の作成

ストリームデータを送信する Kinesis Data Stream を作成します。

このストリームには、以下の様なフォーマットのレコードを送信します。

{
  "user": 2,
  "heartrate": 143,
  "power": 194,
  "cadence": 93,
  "timestamp": "2022-05-20 14:50:09"
}

2. Redshift に Kinesis Data Streams用ポリシーを付与

RedshiftがKinesis Data Streamsにアクセスできるよう、Redshiftクラスターに、以下のポリシーを付与したIAMロールを適用します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "ReadStream",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStreamSummary",
                "kinesis:GetShardIterator",
                "kinesis:GetRecords",
                "kinesis:DescribeStream"
            ],
            "Resource": "arn:aws:kinesis:*:0123456789:stream/*"
        },
        {
            "Sid": "ListStream",
            "Effect": "Allow",
            "Action": [
                "kinesis:ListStreams",
                "kinesis:ListShards"
            ],
            "Resource": "*"
        }
    ]
}

3. Kinesis用外部スキーマの作成

Amazon Redshift Spectrumと同様に、Redshift外のデータベースである Kinesis Data Streams を参照するために、外部スキーマを作成します。

CREATE EXTERNAL SCHEMA schema_one
FROM KINESIS
IAM_ROLE 'arn:aws:iam::0123456789:role/redshift-streaming-role';

IAM_ROLE には、手順2 で適用した IAM ロールの ARN を設定します。

4. マテリアライズド・ビューの作成

Kinesisの基本情報とレコード(Data)だけを取得するマテリアライズド・ビューを作成します。

CREATE MATERIALIZED VIEW view_foo AS
SELECT
  approximatearrivaltimestamp,
  partitionkey,
  shardid,
  sequencenumber,
  JSON_PARSE(from_varbyte(Data, 'utf-8')) as Data
FROM
  schema_one.YOUR_STREAM_NAME

YOUR_STREAM_NAME には 手順1で作成したストリーム名を設定します。

5. マテリアライズド・ビューの更新

KinesisからRedshiftのマテリアライズド・ビューにデータを引っ張ってくるために、ビューをリフレッシュします。

REFRESH MATERIALIZED VIEW view_foo

初回リフレッシュ時には、ストリームに存在するすべてのデータを同期します(ストリームのチェックポイントはTRIM_HORIZON)。以降は差分同期です。

6. マテリアライズド・ビューの参照

ストリームデータを覗いてみます。

select *
from view_foo
limit 5

Kinesis Data Streamsのデータを取得できています。

パーティションキー(partitionkey)やシャードID(shardid)などのKinesisストリームの付加情報はデバッグなどで活用しましょう。

7. 複雑なマテリアライズド・ビューの定義

Kinesis には JSON 形式のレコードが送信されています。 属性ごとに展開したマテリアライズド・ビューを定義します。

CREATE MATERIALIZED VIEW view_foo_extract DISTKEY(2) sortkey(1) AS
SELECT
  approximatearrivaltimestamp,
  json_extract_path_text(from_varbyte(data, 'utf-8'), 'user') :: INT as user_id,
  json_extract_path_text(from_varbyte(data, 'utf-8'), 'timestamp') :: varchar(20) as generated_time,
  json_extract_path_text(from_varbyte(data, 'utf-8'), 'power') :: INT as power,
  json_extract_path_text(from_varbyte(data, 'utf-8'), 'cadence') :: INT as cadence,
  json_extract_path_text(from_varbyte(data, 'utf-8'), 'heartrate') :: INT as heartrate
FROM
  schema_one.foo;

このビューでは、さらに以下を行っています。

• ユーザーID(user_id)でデータ分散
• Kinesisにレコード送信された時刻(approximatearrivaltimestamp)でソート

先程と同じく、ビューをリフレッシュし、レコードを確認します。

REFRESH MATERIALIZED VIEW view_foo_extract

select *
from view_foo_extract
limit 5

過去3分のレコードを対象に ユーザx分単位でデータをサマってみましょう。

select
  user_id,
  to_timestamp(generated_time, 'YYYY-MM-DD HH24:MI') as minute,
  avg(power)
from
  view_foo_extract
where
  approximatearrivaltimestamp > current_timestamp - interval '3 minutes'
group by
  user_id,
  to_timestamp(generated_time, 'YYYY-MM-DD HH24:MI')
order by
  1,
  2

8. ニアリアルタイム処理の確認

ニアリアルタイムに処理できていることを確認するために、マテリアライズド・ビューをリフレッシュ後、過去3分を対象に、Kinesisにレコード送信された時刻(approximatearrivaltimestamp)の最小、最大値を確認します。

SELECT current_timestamp,
       min(approximatearrivaltimestamp),
       max(approximatearrivaltimestamp)
FROM view_foo
WHERE approximatearrivaltimestamp > current_timestamp - interval '3 minutes'

• 現在時刻 17:16:58.891641 に対して、最新のレコードは9秒前の 17:16:49
• 3分前は 17:13:58.891641 に対して、3分以内の最古のレコードはほぼ同時刻の 17:13:59

とニアリアルタイムに処理されています。

ストリームデータはビューをリフレッシュしないと更新されないため、クエリーのスケジュール機能を利用し、マテリアライズド・ビューのリフレッシュを定期実行しましょう。

アーキテクチャー

AWSの機能紹介動画で、アーキテクチャーが紹介されていました。

永続的なデータ向けテーブル(permanent table)とストリームデータ用テーブル(Streaming Table)が別れていますね。

さらに、ストリーム用テーブルには Kinesis Data Streams だけでなく、"Amazon Managed Kafka Service ... others"の文字も見えます。

Redshiftがストリームデータに対応し、その第一弾として Kinesis Data Streams 対応が発表されたとみなせ、今後も続々とストリーム対応が進みそうです。

まとめ

Amazon RedshiftからAmazon Kinesis Data Streamのストリームデータをマテリアライズド・ビュー経由でニアリアルタイムにインジェストする機能がプレビュー提供されています。

データパイプラインの観点からは、Firehose・S3を挟むことで構成が複雑になっていたり、データ処理までのレイテンシーが発生するといった課題が解消されます。

フェデレーテッド・クエリの観点からは、Redshiftが新たにストリームデータにも対応したとみなせます。

本機能はKinesis Data StreamsとシームレスにSQL連携できるため、将来的にはKinesis Data Analyticsが担っていた処理の一部もRedshiftへの移行が進むでしょう。

繰り返しとなりますが、本機能はパブリックプレビューで提供されているため、正式版のリリースまでに機能や仕様は変更される可能性があります。 あくまでも検証目的でご利用ください。

参考

• Amazon Redshift announces public preview of Streaming Ingestion for Kinesis Data Streams
• Streaming ingestion (preview) - Amazon Redshift
• Real-time Analytics with Amazon Redshift Streaming Ingestion -Demo | Amazon Web Services - YouTube