ゲーム分析を Amazon Aurora と Amazon Redshift の Zero-ETL 統合ではじめよう

{
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Action": [
            "rds:CreateIntegration"
        ],
        "Resource": [
            "arn:aws:rds:{region}:{account-id}:cluster:{source-cluster}",
            "arn:aws:rds:{region}:{account-id}:integration:*"
        ]
    },
    {
        "Effect": "Allow",
        "Action": [
            "rds:DescribeIntegration"
        ],
        "Resource": ["*"]
    },
    {
        "Effect": "Allow",
        "Action": [
            "rds:DeleteIntegration"
        ],
        "Resource": [
            "arn:aws:rds:{region}:{account-id}:integration:*"
        ]
    },
    {
        "Effect": "Allow",
        "Action": [
            "redshift:CreateInboundIntegration"
        ],
        "Resource": [
            "arn:aws:redshift-serverless:{region}:{account-id}:namespace/{namespace-uuid}"
        ]
    }]
}

CREATE DATABASE sampledb;

CREATE TABLE sampledb.payments (
    id int AUTO_INCREMENT NOT NULL PRIMARY KEY,
    order_id varchar(255),
    platform_product_id varchar(255),
    purchase_time int,
    purchase_state int
);

CREATE TABLE sampledb.products (
    id int AUTO_INCREMENT NOT NULL PRIMARY KEY,
    platform_product_id varchar(255),
    price int
);

pip install PyMySQL

import pymysql.cursors
import random
import uuid
import time

PAYMENT_RECORDS_NUM = 1000
PURCHASE_STATE_LIST = [
    0, # 決済待ち
    1, # 決済済み
    2  # 上記以外の例外的な決済状態
]
PLATFORM_PRODUCTS = [
    ["item_1", 100],
    ["item_2", 200],
    ["item_3", 300],
    ["item_4", 400]
]

platform_product_ids = list(map(lambda x: x[0], PLATFORM_PRODUCTS))

connection = pymysql.connect(
    host = "{Aurora MySQL ライターエンドポイント}",
    user = "{DB ユーザー名}",
    password = "{DB パスワード}",
    database = "sampledb",
    cursorclass = pymysql.cursors.DictCursor
)

with connection:
    with connection.cursor() as cursor:
        sql = "INSERT INTO products (platform_product_id, price) VALUES (%s, %s)"
        cursor.executemany(sql, PLATFORM_PRODUCTS)

        current_timestamp = int(time.time())
        payments = []
        for i in range(PAYMENT_RECORDS_NUM):
            order_id = uuid.uuid4()
            platform_product_id = random.choice(platform_product_ids)
            purchase_time = random.randint(current_timestamp - 86400, current_timestamp)
            purchase_state = random.choices(PURCHASE_STATE_LIST, [1, 100, 1])
            payments.append([order_id, platform_product_id, purchase_time, purchase_state])
        sql = "INSERT INTO payments (order_id, platform_product_id, purchase_time, purchase_state) VALUES (%s, %s, %s, %s)"
        cursor.executemany(sql, payments)
    connection.commit()

python insert_sample_data.py

SELECT sampledb.products.platform_product_id, SUM(sampledb.products.price)
FROM sampledb.payments, sampledb.products
WHERE sampledb.payments.platform_product_id = sampledb.products.platform_product_id
AND sampledb.payments.purchase_state = 1
GROUP BY sampledb.products.platform_product_id
ORDER BY sampledb.products.platform_product_id;

SELECT
    sampledb.products.platform_product_id,
    SUM(sampledb.products.price),
    DATE_TRUNC('HOUR', TIMESTAMP 'epoch' + sampledb.payments.purchase_time * INTERVAL '1 second') as purchase_hour
FROM sampledb.payments, sampledb.products
WHERE sampledb.payments.platform_product_id = sampledb.products.platform_product_id
AND sampledb.payments.purchase_state = 1
GROUP BY sampledb.products.platform_product_id, purchase_hour
ORDER BY sampledb.products.platform_product_id, purchase_hour;