SQLite3データをData Firehose経由でS3 TablesのIcebergテーブルに格納する

import sqlite3
import random
from datetime import datetime, timedelta

## create database.
with sqlite3.connect("main.db") as con:
    cur = con.cursor()

    ## create tables.
    cur.execute("""
        CREATE TABLE IF NOT EXISTS temperature_humidity_histories (
            sensor_id,
            created_at,
            temperature,
            humidity
        )
    """)

    cur.execute("""
        CREATE TABLE IF NOT EXISTS thermohygrometers (
            sensor_id,
            sensor_name
        )
    """)

    ## insert data
    sensor_cnt = 10
    sensors = [ (i, f'meter{i}',) for i in range(1, sensor_cnt+1) ]
    cur.executemany("""
        INSERT INTO thermohygrometers VALUES (?, ?)
    """, sensors)

    days_cnt = 60
    base_date = datetime.now() - timedelta(days=days_cnt)
    datelist = [
        (base_date + timedelta(days=d)).strftime('%Y-%m-%d %H:%M:%S')
        for d in range(days_cnt)
    ]
    histories = [
        (random.choice(range(1, sensor_cnt+1)),
        date,
        random.normalvariate(20.0, 2),
        random.normalvariate(40.0, 10),)
        for date in datelist
    ]
    cur.executemany("""
        INSERT INTO temperature_humidity_histories VALUES (?, ?, ?, ?)
    """, histories)

    con.commit()

import { Bucket } from 'aws-cdk-lib/aws-s3';
import { CfnTableBucket } from 'aws-cdk-lib/aws-s3tables';

    // S3 Bucket
    // ソース用バケット
    this.srcBucket = new Bucket(this, 'SourceBucket', {
      removalPolicy: cdk.RemovalPolicy.RETAIN,
      eventBridgeEnabled: true,   // EventBridgeへのイベント通知を有効化
    });

    // ストリーム用バケット (配信失敗時のバックアップデータ格納先)
    this.deliveryStreamBucket = new Bucket(this, 'DeliveryStreamBucket', {
      removalPolicy: cdk.RemovalPolicy.RETAIN,
    });

    // S3 Table Bucket
    // ターゲットIcebergテーブルバケット
    this.targetTableBucket = new CfnTableBucket(this, 'TableBucket', {
      tableBucketName: 'firehose-s3table',
    });

## テーブルバケットのカタログに対する権限追加
aws lakeformation grant-permissions \
  --catalog-id "${AWS_ACCOUNT_ID}" \
  --principal "{\"DataLakePrincipalIdentifier\": \"${AWS_PRINCIPAL_ARN}\"}" \
  --resource "{\"Catalog\": {\"Id\": \"${AWS_ACCOUNT_ID}:s3tablescatalog/${TABLE_BUCKET_NAME}\"}}" \
  --permissions "ALL" \
  --permissions-with-grant-option "ALL"

## namespace (database) 作成
## namespace名は 'main' としています。
aws s3tables create-namespace \
  --table-bucket-arn ${TABLE_BUCKET_ARN} \
  --namespace main

## テーブルバケットのnamespace (database)に対する権限追加
aws lakeformation grant-permissions \
  --catalog-id "${AWS_ACCOUNT_ID}" \
  --principal "{\"DataLakePrincipalIdentifier\": \"${AWS_PRINCIPAL_ARN}\"}" \
  --resource "{\"Database\": { \"CatalogId\": \"${AWS_ACCOUNT_ID}:s3tablescatalog/${TABLE_BUCKET_NAME}\", \"Name\": \"main\" }}" \
  --permissions "ALL" \
  --permissions-with-grant-option "ALL"

## テーブル作成
## テーブル名は 'temperature_humidity_histories' としています。
aws s3tables create-table \
  --table-bucket-arn ${TABLE_BUCKET_ARN} \
  --namespace main \
  --name temperature_humidity_histories \
  --format 'ICEBERG' \
  --metadata '{"iceberg":{"schema":{"fields":[{"name":"sensor_id","type":"int","required":true},{"name":"created_at","type":"timestamp"},{"name":"temperature","type":"double"},{"name":"humidity","type":"double"}]}}}'

import { Role, ServicePrincipal, ManagedPolicy, PolicyStatement } from 'aws-cdk-lib/aws-iam';

    // IAM Role
    // Lambda 関数用ロール
    const loadFunctionRole = new Role(this, 'LoadFunctionRole', {
      assumedBy: new ServicePrincipal('lambda.amazonaws.com'),
      managedPolicies: [
        ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSLambdaBasicExecutionRole'),
        ManagedPolicy.fromAwsManagedPolicyName('AmazonS3FullAccess'),
        ManagedPolicy.fromAwsManagedPolicyName('AmazonKinesisFirehoseFullAccess'),
      ],
    });

    // Data Firehose 配信ストリーム用ロール
    const deliveryStreamIcebergRole = new Role(this, 'DeliveryStreamIcebergRole', {
      assumedBy: new ServicePrincipal('firehose.amazonaws.com'),
    });
    deliveryStreamIcebergRole.addToPolicy(new PolicyStatement({
      actions: [
        "glue:GetTable",
        "glue:GetDatabase",
        "glue:UpdateTable",
      ],
      resources: ["*"],
    }));
    deliveryStreamIcebergRole.addToPolicy(new PolicyStatement({
      actions: [
        "lakeformation:GetDataAccess",
      ],
      resources: [
        "*",
      ],
    }));
    deliveryStreamIcebergRole.addToPolicy(new PolicyStatement({
      actions: [
        "logs:PutLogEvents",
      ],
      resources: [
        "*"
      ],
    }));

    // Data Firehose の配信失敗時のバックアップ先S3にアクセスするロール
    const deliveryStreamS3Role = new Role(this, 'DeleveryStreamS3Role', {
      assumedBy: new ServicePrincipal('firehose.amazonaws.com'),
    });
    deliveryStreamS3Role.addToPolicy(new PolicyStatement({
      actions: [
        "s3:AbortMultipartUpload",
        "s3:GetBucketLocation",
        "s3:GetObject",
        "s3:ListBucket",
        "s3:ListBucketMultipartUploads",
        "s3:PutObject",
      ],
      resources: [
        `${props?.deliveryStreamBucket.bucketArn}`,
        `${props?.deliveryStreamBucket.bucketArn}/*`,
      ],
    }));
    deliveryStreamS3Role.addToPolicy(new PolicyStatement({
      actions: [
        "logs:PutLogEvents",
      ],
      resources: [
        "*"
      ],
    }));

import { CfnDatabase } from 'aws-cdk-lib/aws-glue';

    // Resource Link for S3 Tables Database
    const s3tablesResourceLink = new CfnDatabase(this, 'S3TablesResourceLink', {
      catalogId: `${props?.env?.account}`,
      databaseInput: {
        name: dbResourceLinkName,
        targetDatabase: {
          catalogId: `${props?.env?.account}:s3tablescatalog/${props?.targetTableBucket.tableBucketName}`,
          databaseName: databaseName,
        },
      },
    });

import { CfnPrincipalPermissions } from 'aws-cdk-lib/aws-lakeformation';

    // Lake Formation
    new CfnPrincipalPermissions(this, 'ResourceLinkPermission', {
      principal: {
        dataLakePrincipalIdentifier: deliveryStreamIcebergRole.roleArn,
      },
      resource: {   // リソースリンク(DB)に対する権限付与
        database: {
          catalogId: s3tablesResourceLink.catalogId,
          name: dbResourceLinkName,
        },
      },
      permissions: ["DESCRIBE"],
      permissionsWithGrantOption: [],
    });

## データベースリソースリンク先のテーブルへの権限付与
## AWS_PRINCIPAL_ARN は、Data Firehoseの配信ストリームにアタッチするIAMロールのARN
aws lakeformation grant-permissions \
  --catalog-id "${AWS_ACCOUNT_ID}" \
  --principal "{\"DataLakePrincipalIdentifier\": \"${AWS_PRINCIPAL_ARN}\"}" \
  --resource "{\"Table\": {\"CatalogId\": \"${AWS_ACCOUNT_ID}\", \"DatabaseName\": \"resource_link_to_s3tables_main\", \"TableWildcard\": {} }}" \
  --permissions "ALL"

## 実際のS3Tablesのテーブルへの権限付与
aws lakeformation grant-permissions \
  --catalog-id "${AWS_ACCOUNT_ID}" \
  --principal "{\"DataLakePrincipalIdentifier\": \"${AWS_PRINCIPAL_ARN}\"}" \
  --resource "{\"Table\": {\"CatalogId\": \"${AWS_ACCOUNT_ID}:s3tablescatalog/${TABLE_BUCKET_NAME}\", \"DatabaseName\": \"main\", \"TableWildcard\": {} }}" \
  --permissions "ALL"

import { CfnDeliveryStream } from 'aws-cdk-lib/aws-kinesisfirehose';

    // Data Firehose
    const deliveryStream = new CfnDeliveryStream(this, 'DeliveryStream', {
      deliveryStreamType: 'DirectPut',
      icebergDestinationConfiguration: {
        roleArn: deliveryStreamIcebergRole.roleArn,
        bufferingHints: {
          intervalInSeconds: 30,    // 開発向けに早くデータを反映してもらうため
          sizeInMBs: 5,
        },
        retryOptions: {
          durationInSeconds: 30,    // 再試行時間も短めにした
        },
        catalogConfiguration: {
          catalogArn: `arn:aws:glue:${props?.env?.region}:${props?.env?.account}:catalog`,
        },
        destinationTableConfigurationList: [{
          destinationDatabaseName: dbResourceLinkName,
          destinationTableName: tableName,
          uniqueKeys: ['sensor_id', 'created_at'],    // UpdateやDeleteの際に使われるが、今回はInsertしかしないので無視される
          s3ErrorOutputPrefix: `dstErrIceberg/${dbResourceLinkName}/${tableName}`,
        }],
        s3Configuration: {
          // 配信エラー発生時のバックアップ先S3バケットの設定
          bucketArn: `${props?.deliveryStreamBucket.bucketArn}`,
          roleArn: deliveryStreamS3Role.roleArn,
          cloudWatchLoggingOptions: {
            // バックアップエラー発生時のCloudWatch Logs設定
            enabled: true,
            logGroupName: backupLogGroup.logGroupName,
            logStreamName: backupLogStream.logStreamName,
          },
        },
        cloudWatchLoggingOptions: {
          // 配信エラー発生時のCloudWatch Logs設定
          enabled: true,
          logGroupName: streamLogGroup.logGroupName,
          logStreamName: streamLogStream.logStreamName,
        },
      },
    });

import os
import shutil
import sqlite3
import boto3
import botocore
import json
from datetime import datetime
from pprint import pprint

def handler(event, _):
    if event.get("Records"):
        ## S3イベント通知
        src_bucket = event["Records"][0]["s3"]["bucket"]["name"]
        src_objkey = event["Records"][0]["s3"]["object"]["key"]
    else:
        ## EventBridgeイベント通知
        src_bucket = event["detail"]["bucket"]["name"]
        src_objkey = event["detail"]["object"]["key"]

    delivery_stream = os.environ.get('TGT_STREAM')
    target_table = os.environ.get('TGT_TABLE')

    s3_client = boto3.client('s3')
    firehose_client = boto3.client('firehose')

    ## download sqlite database file.
    tmp_dir = '/tmp/load'
    local_path = os.path.join(tmp_dir, os.path.dirname(src_objkey))
    if not os.path.exists(local_path):
        os.makedirs(local_path)
    local_objkey = os.path.join(local_path, os.path.basename(src_objkey))
    s3_client.download_file(src_bucket, src_objkey, local_objkey)

    with sqlite3.connect(local_objkey) as con:
        cur = con.cursor()

        ## scan "temperature_humidity_histories" table data
        batch_size = 10
        cur.execute(f"""
            SELECT sensor_id,
                   created_at,
                   temperature,
                   humidity
            FROM {target_table}
        """)
        rows = cur.fetchmany(batch_size)
        while (rows is not None) and rows:
            data = [ {
                "sensor_id": row[0],
                "created_at": datetime.strptime(row[1], '%Y-%m-%d %H:%M:%S').strftime('%Y-%m-%dT%H:%M:%SZ'),
                "temperature": row[2],
                "humidity": row[3] } for row in rows ]
            records = [ { "Data": json.dumps(record).encode() } for record in data ]
            res = firehose_client.put_record_batch(
                DeliveryStreamName=delivery_stream,
                Records=records,
            )
            print("\nstream response:")
            pprint(res)
            rows = cur.fetchmany(batch_size)

        cur.close()

    shutil.rmtree(local_path)
    return {
        "src_bucket": src_bucket,
    }

import { Function, Runtime, Code } from 'aws-cdk-lib/aws-lambda';
import { Rule } from 'aws-cdk-lib/aws-events';
import { LambdaFunction as targetLambdaFunction } from 'aws-cdk-lib/aws-events-targets';

    // Lambda Function
    const loadFunction = new Function(this, 'LoadFunction', {
      runtime: Runtime.PYTHON_3_12,
      code: Code.fromAsset('lambda/load'),
      handler: 'lambda_function.handler',
      memorySize: 256,
      timeout: cdk.Duration.seconds(10),
      role: loadFunctionRole,
      environment: {
        SRC_BUCKET: `${props?.sourceBucket.bucketName}`,
        TGT_STREAM: deliveryStream.ref,
        TGT_TABLE: tableName,
      },
    });

    // EventBridge Rule
    const onCreateS3Object = new Rule(this, 'onCreateS3ObjectRule', {
      eventPattern: {
        source: ["aws.s3"],
        detailType: ["Object Created"],
        detail: {
          bucket: {
            name: [props?.sourceBucket.bucketName]
          },
        },
      },
      targets: [
        new targetLambdaFunction(loadFunction, {}),
      ]
    });

select *
from temperature_humidity_histories
;