やること
※以下記事の続きです。
Redshift Spectrumを利用してS3から差分データ（追加・更新・削除）を抽出し、Redshiftテーブルを更新する。
（EC2から操作します。）
前提
S3に配置した以下2基準データに対し、SQLを使用して差分抽出、Redshiftテーブルの更新を行います。
- today.csv
1,ALEX,23
2,HANNA,24
3,TARO,30
6,JHON,28
7,JOJI,29
- yesterday.csv
1,ALEX,23
2,HANNA,24
3,TARO,25
4,HANAKO,26
5,STEVE,27
※Redshiftテーブルはyesterday.csvと同じ内容（昨日基準）。
today.csvとyesterday.csvの比較し差分を抽出⇒テーブル更新することで最新化する。
（イメージ）
【差分データ（追加・更新・削除）を抽出するイメージ】
（追加データ：today.csvのみに存在）
6,JHON,28
7,JOJI,29
（削除データ：yesetrday.csvのみに存在）
4,HANAKO,26
5,STEVE,27
（更新データ：today.csvとyesterday.csvにともに存在かつ更新された箇所がある）
3,TARO,25 ⇒ 3,TARO,30
手順
①Redshiftスキーマ・テーブル作成（準備）
②S3から追加レコード抽出⇒Redshiftテーブルにロード
③S3から削除レコード抽出⇒Redshiftテーブルから対象レコード削除
④S3から更新レコード抽出⇒Redshiftテーブルの対象レコード更新
①Redshiftスキーマ・テーブル作成（準備）
以下DDLでRedshiftにtestスキーマとtestテーブルを作成します。
【スキーマ作成】
CREATE SCHEMA test;
【テーブル作成】
CREATE TABLE test.test(id INTEGER primary key,
name VARCHAR(50) not null,
age INTEGER not null);
実行結果
dev=# CREATE SCHEMA test;
CREATE SCHEMA
dev=# CREATE TABLE test.test(id INTEGER primary key,
name VARCHAR(50) not null,
age INTEGER not null);
CREATE TABLE
作成が完了したら以下コマンドで確認します。
（\dn：スキーマ情報取得）
dev=# \dn
List of schemas
name | owner
-----------------+-------
catalog_history | rdsdb
public | rdsdb
spectrum_test | admin
test | admin
(4 rows)
（SET search_path TO スキーマ名：接続先スキーマ切り替え）
dev=# SET search_path TO test;
SET
（\d：接続先スキーマに紐づくテーブル情報確認）
dev=# \d
List of relations
schema | name | type | owner
--------+------+-------+-------
test | test | table | admin
(1 row)
また、yesterday.csvと同じ内容のデータをINSERTします。
BEGIN;
TRUNCATE TABLE test.test;
INSERT INTO test.test VALUES(1,'ALEX',23);
INSERT INTO test.test VALUES(2,'HANNA',24);
INSERT INTO test.test VALUES(3,'TARO',25);
INSERT INTO test.test VALUES(4,'HANAKO',26);
INSERT INTO test.test VALUES(5,'STEVE',27);
COMMIT;
実行結果
dev=# BEGIN;
TRUNCATE TABLE test.test;
INSERT INTO test.test VALUES(1,'ALEX',23);
INSERT INTO test.test VALUES(2,'HANNA',24);
INSERT INTO test.test VALUES(3,'TARO',25);
INSERT INTO test.test VALUES(4,'HANAKO',26);
INSERT INTO test.test VALUES(5,'STEVE',27);
COMMIT;
BEGIN
TRUNCATE TABLE and COMMIT TRANSACTION
INSERT 0 1
INSERT 0 1
INSERT 0 1
INSERT 0 1
INSERT 0 1
COMMIT
完了したらSELECT文で確認します。
dev=# SELECT * FROM test.test;
id | name | age
----+--------+-----
1 | ALEX | 23
2 | HANNA | 24
3 | TARO | 25
4 | HANAKO | 26
5 | STEVE | 27
(5 rows)
以上でスキーマ・テーブル作成は完了です。
②S3から追加レコード抽出⇒Redshiftテーブルにロード
ここからは前日基準のCSVファイルと当日基準のCSVを比較して差分データを抽出し、Redshiftテーブルを更新します。
まずは、新規追加されたレコードの抽出・Redshiftへの新規追加を行います。
新規追加レコードの抽出は、前日基準と当日基準を比較し、当日基準にのみ存在するレコードを抽出する方式で行います。
図式化すると以下のようなイメージです。
また、2次元表で表すと以下のようなイメージです。
上記抽出は、前日基準に同じid（主キー）を持つ行が存在しないもののみを当日基準から抽出することで実現できます。
今回はNOT EXISTSを用いて実装します。
SELECT文は以下です。
SELECT * FROM spectrum_test.today t
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.yesterday y
WHERE t.id = y.id
);
実行結果は以下の通りです。
dev=# SELECT * FROM spectrum_test.today t
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.yesterday y
WHERE t.id = y.id
);
id | name | age
----+------+-----
6 | JHON | 28
7 | JOJI | 29
(2 rows)
想定通りのデータを抽出することができました！
抽出したデータはRedshiftのテーブル（test.test）に全量INSERTします。
全量INSERT後のテーブルは以下のようになります。
最終的なSQLは以下の通りです。
INSERT INTO test.test
SELECT * FROM spectrum_test.today t
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.yesterday y
WHERE t.id = y.id
);
実行結果は以下の通りです。
dev=# INSERT INTO test.test
SELECT * FROM spectrum_test.today t
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.yesterday y
WHERE t.id = y.id
);
INSERT 0 2
test.testのデータをSELECTして結果を確認します。
dev=# SELECT * FROM test.test ORDER BY id;
id | name | age
----+--------+-----
1 | ALEX | 23
2 | HANNA | 24
3 | TARO | 25
4 | HANAKO | 26
5 | STEVE | 27
6 | JHON | 28
7 | JOJI | 29
(7 rows)
想定通りのテーブルの状態になりました！
③S3から削除レコード抽出⇒Redshiftテーブルから対象レコード削除
次に、削除されたレコードの抽出・Redshiftからの削除を行います。
削除レコードの抽出は、新規追加の逆です。
前日基準と当日基準を比較し、前日基準にのみ存在するレコードを抽出する方式で行います。
図式化すると以下のようなイメージです。
また、2次元表で表すと以下のようなイメージです。
上記抽出は、当日基準に同じid（主キー）を持つ行が存在しないもののみを前日基準から抽出することで実現できます。
新規追加レコード抽出と同様、NOT EXISTSを用いて実装します。
SELECT文は以下です。
SELECT * FROM spectrum_test.yesterday y
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.today t
WHERE t.id = y.id
);
実行結果は以下の通りです。
dev=# SELECT * FROM spectrum_test.yesterday y
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.today t
WHERE t.id = y.id
);
id | name | age
----+--------+-----
4 | HANAKO | 26
5 | STEVE | 27
(2 rows)
想定通りのデータを抽出することができました！
抽出したデータはRedshiftのテーブル（test.test）から全量DELETEします。
全量DELETE後のテーブルは以下のようになります。
今回は、テーブルを結合しながら削除するために使うPostgreSQL拡張のUSINGを使用します。
最終的なSQLは以下の通りです。
DELETE FROM test.test
USING (
SELECT * FROM spectrum_test.yesterday y
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.today t
WHERE t.id = y.id
)
) d
WHERE test.test.id = d.id;
実行結果は以下の通りです。
dev=# DELETE FROM test.test
USING (
SELECT * FROM spectrum_test.yesterday y
WHERE NOT EXISTS (
SELECT * FROM spectrum_test.today t
WHERE t.id = y.id
)
) d
WHERE test.test.id = d.id;
DELETE 2
test.testのデータをSELECTして結果を確認します。
dev=# SELECT * FROM test.test ORDER BY id;
id | name | age
----+-------+-----
1 | ALEX | 23
2 | HANNA | 24
3 | TARO | 25
6 | JHON | 28
7 | JOJI | 29
(5 rows)
想定通りのテーブルの状態になりました！
④S3から更新レコード抽出⇒Redshiftテーブルの対象レコード更新
最後に、更新されたレコードの抽出・Redshiftテーブルの更新を行います。
更新レコードの抽出は、当日基準と前日基準に同じid（主キー）が存在し、かつnameとageのどちらかが一致しないレコードを抽出する方式で行います。
図式化すると以下のようなイメージです。
また、2次元表で表すと以下のようなイメージです。
上記抽出は、当日基準と前日基準をid（主キー）で内部結合し、name／age（主キー以外の比較カラム）で不一致をのもの抽出することで実現できます。
比較カラムの不一致レコードの抽出はIS DISTINCT FROMを用いて実装します。
※IS DISTINCT FROM：PostgreSQL特有の比較演算子であり、NULLを安全に扱うことができます。
a IS DISTINCT FROM bと記述することで、aもしくはbがNULLかどうかにかかわらず確実に不一致であるものを抽出します。
SELECT文は以下です。
SELECT t.id, t.name, t.age
FROM spectrum_test.today t
INNER JOIN spectrum_test.yesterday y
ON t.id = y.id
WHERE
t.name IS DISTINCT FROM y.name
OR
t.age IS DISTINCT FROM y.age;
また、IS DISTINCT FROMを以下のように記述しても上記と同じ意味になります。
WHERE (t.name, t.age) IS DISTINCT FROM (y.name, y.age);
実行結果は以下の通りです。
dev=# SELECT t.id, t.name, t.age
FROM spectrum_test.today t
INNER JOIN spectrum_test.yesterday y
ON t.id = y.id
WHERE (t.name, t.age) IS DISTINCT FROM (y.name, y.age);
id | name | age
----+------+-----
3 | TARO | 30
(1 row)
想定通りのデータを抽出することができました！
抽出したデータはRedshiftのテーブル（test.test）の該当レコードを全量UPDATEします。
全量UPDATE後のテーブルは以下のようになります。
PostgreSQLにおける、他テーブルの抽出結果を使って更新するUPDATE...SET...FROM構文を使います。
最終的なSQLは以下の通りです。
UPDATE test.test rt
SET
id = u.id,
name = u.name,
age = u.age
FROM (
SELECT t.id, t.name, t.age
FROM spectrum_test.today t
INNER JOIN spectrum_test.yesterday y
ON t.id = y.id
WHERE (t.name, t.age) IS DISTINCT FROM (y.name, y.age)
) u
WHERE rt.id = u.id;
実行結果は以下の通りです。
dev=# UPDATE test.test rt
SET
id = u.id,
name = u.name,
age = u.age
FROM (
SELECT t.id, t.name, t.age
FROM spectrum_test.today t
INNER JOIN spectrum_test.yesterday y
ON t.id = y.id
WHERE (t.name, t.age) IS DISTINCT FROM (y.name, y.age)
) u
WHERE rt.id = u.id;
UPDATE 1
test.testのデータをSELECTして結果を確認します。
dev=# SELECT * FROM test.test ORDER BY id;
id | name | age
----+-------+-----
1 | ALEX | 23
2 | HANNA | 24
3 | TARO | 30
6 | JHON | 28
7 | JOJI | 29
(5 rows)
想定通りのテーブルの状態になりました！
参考にさせていただいた記事