Amazon Kinesis Data Streams および AWS Lambda を使用して、Amazon RDS for PostgreSQL の変更をストリーミングする

postgres> show wal_level;
+-------------+
| wal_level   |
|-------------|
| logical     |
+-------------+
SHOW
Time: 0.002s

postgres> show max_replication_slots;
+-------------------------+
| max_replication_slots   |
|-------------------------|
| 5                       |
+-------------------------+
SHOW
Time: 0.002s

postgres> create user repluser password 'replpass';
CREATE ROLE
Time: 0.003s

postgres> grant rds_replication to repluser;
GRANT ROLE
Time: 0.002s

aws kinesis create-stream --stream-name Foo --shard-count 1

[ec2-user@ip-172-xx-yy-zzz ~]$ cat ~/.aws/credentials 
[default]
aws_access_key_id = XXXXXXXXXXXXXXX
aws_secret_access_key = tSQYYYYYYXXX+TrDYYYYYYYY

import boto3
import json
import random
import calendar
import time
from datetime import datetime
import psycopg2
from psycopg2.extras import LogicalReplicationConnection

my_stream_name = 'Foo'
kinesis_client = boto3.client('kinesis', region_name='us-east-1')
my_connection  = psycopg2.connect(
                   "dbname='postgres' host='mypgdb.abcdefghijk.us-east-1.rds.amazonaws.com' user='repluser' password='replpass'" ,
                   connection_factory = LogicalReplicationConnection)
cur = my_connection.cursor()
cur.drop_replication_slot('wal2json_test_slot')
cur.create_replication_slot('wal2json_test_slot', output_plugin = 'wal2json')
cur.start_replication(slot_name = 'wal2json_test_slot', options = {'pretty-print' : 1}, decode= True)

def consume(msg):
    kinesis_client.put_record(StreamName=my_stream_name, Data=json.dumps(msg.payload), PartitionKey="default")
    print (msg.payload)

cur.consume_stream(consume)

postgres> \d+ employee
+----------+---------+-------------+-----------+----------------+---------------+
| Column   | Type    | Modifiers   | Storage   | Stats target   | Description   |
|----------+---------+-------------+-----------+----------------+---------------|
| empno    | integer |  not null   | plain     | <null>         | <null>        |
| deptid   | integer |             | plain     | <null>         | <null>        |
+----------+---------+-------------+-----------+----------------+---------------+
Indexes:
    "employee_pkey" PRIMARY KEY, btree (empno)
Foreign-key constraints:
    "employee_deptid_fkey" FOREIGN KEY (deptid) REFERENCES dept(deptid)
Has OIDs: no

Time: 0.032s

postgres> insert into employee values (1057,100);
INSERT 0 1
Time: 0.002s
postgres> insert into employee values (1058,100);
INSERT 0 1
Time: 0.002s
postgres> update employee set empno = 1059 where empno = 1057;
UPDATE 1
Time: 0.002s

[ec2-user@ip-172-31-63-237 ~]$ python kinesis_producer_from_pg.py
{
    "change": [
        {
            "kind": "insert",
            "schema": "jatin",
            "table": "employee",
            "columnnames": ["empno", "deptid"],
            "columntypes": ["integer", "integer"],
            "columnvalues": [1057, 100]
        }
    ]
}
{
    "change": [
        {
            "kind": "insert",
            "schema": "jatin",
            "table": "employee",
            "columnnames": ["empno", "deptid"],
            "columntypes": ["integer", "integer"],
            "columnvalues": [1058, 100]
        }
    ]
}
{
    "change": [
        {
            "kind": "update",
            "schema": "jatin",
            "table": "employee",
            "columnnames": ["empno", "deptid"],
            "columntypes": ["integer", "integer"],
            "columnvalues": [1059, 100],
            "oldkeys": {
                "keynames": ["empno"],
                "keytypes": ["integer"],
                "keyvalues": [1057]
            }
    }
    ]
}

ec2-user@ip-172-xx-yy-zzz $ ls -ltr
drwxrwxr-x 2 ec2-user ec2-user   4096 May 14 00:53 psycopg2
-rw-rw-r-- 1 ec2-user ec2-user 472722 May 19 20:13 mypackage.zip
-rw-rw-r-- 1 ec2-user ec2-user   1350 May 19 20:36 app.py

import sys
import logging
import psycopg2
import boto3
import json
import random
import calendar
import time
from datetime import datetime
from psycopg2.extras import LogicalReplicationConnection

my_stream_name = 'Foo'
kinesis_client = boto3.client('kinesis', region_name='us-east-1')

logger = logging.getLogger()
logger.setLevel(logging.INFO)

try:
    my_connection  = psycopg2.connect(
                      "dbname='postgres' host='mypgdb.xxxxxxxxxxxxx.us-east-1.rds.amazonaws.com' user='repluser' password='replpass'" ,
                      connection_factory = LogicalReplicationConnection)
except:
    logger.error("ERROR: Unexpected error: Could not connect to RDS for PostgreSQL instance.")
    sys.exit()

logger.info("SUCCESS: Connection to RDS for PostgreSQL instance succeeded")

def handler(event, context):
    """
    この関数は、コンテンツを RDS for PostgreSQL から Kinesis へストリーミングする
    """

    cur = my_connection.cursor()
    cur.create_replication_slot('wal2json_test_slot', output_plugin = 'wal2json')
    cur.start_replication(slot_name = 'wal2json_test_slot', options = {'pretty-print' : 1}, decode= True)

    cur.consume_stream(consume)

def consume(msg):
    kinesis_client.put_record(StreamName=my_stream_name, Data=json.dumps(msg.payload), PartitionKey="default")
    print (msg.payload)

aws lambda create-function \
--region us-east-1 \
--function-name postgres-kinesis \
--zip-file fileb://mypackage.zip \
--role arn:aws:iam::123456789111:role/lambda-role-mysql \
--handler app.handler \
--runtime python2.7 \
--timeout 15 \
--memory-size 512

streamname=Foo; aws kinesis describe-stream --stream-name $streamname --output text | grep SHARDS | awk '{print $2}' | while read shard; do aws kinesis get-shard-iterator --stream-name $streamname --shard-id $shard --shard-iterator-type LATEST --output text | while read iterator; do while output=`aws kinesis get-records --shard-iterator $iterator --output text`; do iterator=`echo "$output" | head -n1 | awk '{print $2}'`; echo "$output" | awk 'NR > 1' | grep RECORDS | while read record; do echo $record | awk '{print $3}' | base64 -id; done; done; done; done

"{\n\t\"change\": [\n\t\t{\n\t\t\t\"kind\": \"insert\",\n\t\t\t\"schema\": \"<>\",\n\t\t\t\"table\": \"employee\",\n\t\t\t\"columnnames\": [\"empno\", \"deptid\"],\n\t\t\t\"columntypes\": [\"integer\", \"integer\"],\n\t\t\t\"columnvalues\": [1234, 100]\n\t\t}\n\t]\n}""{\n\t\"change\": [\n\t\t{\n\t\t\t\"kind\": \"insert\",\n\t\t\t\"schema\": \"<>\",\n\t\t\t\"table\": \"employee\",\n\t\t\t\"columnnames\": [\"empno\", \"deptid\"],\n\t\t\t\"columntypes\": [\"integer\", \"integer\"],\n\t\t\t\"columnvalues\": [2222, 101]\n\t\t}\n\t]\n}""{\n\t\"change\": [\n\t\t{\n\t\t\t\"kind\": \"update\",\n\t\t\t\"schema\": \"<>\",\n\t\t\t\"table\": \"employee\",\n\t\t\t\"columnnames\": [\"empno\", \"deptid\"],\n\t\t\t\"columntypes\": [\"integer\", \"integer\"],\n\t\t\t\"columnvalues\": [3333, 101],\n\t\t\t\"oldkeys\": {\n\t\t\t\t\"keynames\": [\"empno\"],\n\t\t\t\t\"keytypes\": [\"integer\"],\n\t\t\t\t\"keyvalues\": [2222]\n\t\t\t}\n\t\t}\n\t]\n}""{\n\t\"change\": [\n\t\t{\n\t\t\t\"kind\": \"delete\",\n\t\t\t\"schema\": \"<>\",\n\t\t\t\"table\": \"employee\",\n\t\t\t\"oldkeys\": {\n\t\t\t\t\"keynames\": [\"empno\"],\n\t\t\t\t\"keytypes\": [\"integer\"],\n\t\t\t\t\"keyvalues\": [1234]\n\t\t\t}\n\t\t}\n\t]\n}"