Airbyteの最新分と過去分のSCDテーブルをいい感じに統合する

# はじめに
:::note warn
最新のAirbyteの[Incremental Sync - Append + Deduped](https://docs.airbyte.com/using-airbyte/core-concepts/sync-modes/incremental-append-deduped) の同期方法において、本記事で紹介するSCDテーブルは生成されません。

これはDestination V2で廃止された中間テーブルです。
詳しくは下記の記事をご覧ください。
https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2
:::

データ連携業務において、使用しているSaaSからのデータ取得でAirbyteを用いている。
以下の記事に背景などあるので、参照されたい。

https://techblog.zozo.com/entry/airbyte-on-gke

# 問題
**「毎日のデータ連携は増分同期だけど、たまに全量転送したい。
　過去の追跡をしているSCDテーブルは残して、いい感じに統合したい」**

というユースケース。

API経由のデータ連携は何かと問題が起きがち。
APIのRate Limitであったり、エラーハンドリング、増分同期の際はCursor Filed起因によるデータ欠損など...

毎回全量転送（[Full Refresh - Overwrite](https://docs.airbyte.com/using-airbyte/core-concepts/sync-modes/full-refresh-overwrite)）で行えればいいんだけど、データ量が大きかったり、SaaS側のデータ保持期間が決まっていたりすると、そうもいかない。

コネクタのバージョンアップなどで、データ欠損のバグが解消された時に、一度、Cursor Filed指定なしで全量データ転送行いたい時がある。
（Cursor Filed起因による取りこぼしデータが復活するので）

この時、定常運用しているSCDテーブルと上手く統合することで、過去の追跡データも保持したいという場合があったので、紹介する。

# 解決方法
以下の記事で紹介されている例を用いる。

https://airbyte.com/tutorials/incremental-data-synchronization

下記のテーブル転送した時、Destination先で生成されるSCDテーブルと重複排除テーブルを対象にする。
```sql
CREATE TABLE table_one(
  id integer PRIMARY KEY,
  name VARCHAR(200),
  updated_at timestamp DEFAULT NOW() NOT NULL
);
```

以下のクエリを実行すると、新しいSCDテーブルと古いSCDテーブルを統合できる。
```sql
WITH input_data as (
    select "_airbyte_table_two_hashid",
      "id",
      "name",
      "updated_at",
      "_airbyte_ab_id",
      "_airbyte_emitted_at",
      "_airbyte_normalized_at" 
    from new_table_one_scd
    union all
    select "_airbyte_table_two_hashid",
      "id",
      "name",
      "updated_at",
      "_airbyte_ab_id",
      "_airbyte_emitted_at",
      "_airbyte_normalized_at" 
    from old_table_one_scd
),

scd_data as (
    select
      md5(cast(coalesce(cast("id" as text), '') as text)) as _airbyte_unique_key,
      "id",
      "name",
      updated_at,
      updated_at as _airbyte_start_at,
      lag(updated_at) over (
        partition by cast("id" as text)
        order by
            updated_at is null asc,
            updated_at desc,
            _airbyte_emitted_at desc
      ) as _airbyte_end_at,
      case when row_number() over (
        partition by cast("id" as text)
        order by
            updated_at is null asc,
            updated_at desc,
            _airbyte_emitted_at desc
      ) = 1 then 1 else 0 end as _airbyte_active_row,
      _airbyte_ab_id,
      _airbyte_emitted_at,
      _airbyte_table_two_hashid
    from input_data
),

dedup_data as (
    select
        row_number() over (
            partition by
                _airbyte_unique_key,
                _airbyte_start_at,
                _airbyte_emitted_at
            order by _airbyte_active_row desc, _airbyte_ab_id
        ) as _airbyte_row_num,
        md5(cast(coalesce(cast(_airbyte_unique_key as text), '') || '-' || coalesce(cast(_airbyte_start_at as text), '') || '-' || coalesce(cast(_airbyte_emitted_at as text), '') as text)) as _airbyte_unique_key_scd,
        scd_data.*
    from scd_data
)

select
    _airbyte_unique_key,
    _airbyte_unique_key_scd,
    "id",
    "name",
    updated_at,
    _airbyte_start_at,
    _airbyte_end_at,
    _airbyte_active_row,
    _airbyte_ab_id,
    _airbyte_emitted_at,
    now() as _airbyte_normalized_at,
    _airbyte_table_two_hashid
from dedup_data where _airbyte_row_num = 1
  );
```
ここで、new_table_one_scdが新しいSCDテーブル（今回だと全量転送時に作られたテーブル）で、old_table_one_scdが統合させたい古いSCDテーブル。

元記事と比較すると分かると思うが、最初のUNION ALLするテーブルが違うだけで、その後の処理は同じ。
この記事読んで着想を得た。

ちなみに重複排除テーブル（最終的なテーブル）も統合する場合はシンプルで、以下のクエリで統合できる。
```sql
WITH new_table_one_pk_list AS(
    SELECT Id FROM new_table_one
),

old_table_only AS(
    SELECT * FROM old_table_one
    WHERE
        ID NOT IN(SELECT Id FROM new_table_one_pk_list)
)

SELECT * FROM new_table_one
UNION ALL
SELECT * FROM old_table_only
```

# Airbyte固有のカラムについて
上記の処理を考えるにあたって、各テーブルでAirbyte固有のカラムが生成過程を理解する必要があった。
主に、下記の記事の訳と抜粋で紹介する。

https://airbyte.com/tutorials/incremental-data-synchronization

## 重複排除テーブル（最終的なテーブル）
#### `_airbyte_ab_id`
テーブルにおいて一意の値。
データが変更されていない場合でも、Airbyteで同期が行われるたびに別の値が入る。

#### `_airbyte_emitted_at`
転送時の日時。

#### `_airbyte_normalized_at`
生データ(基本的にjson)から正規化された日時。

#### `_airbyte_table_one_hashid`
正規化時のハッシュ化された値。

```sql
md5(cast(coalesce(cast("id" as text), '') || '-' || coalesce(cast("name" as text), '') || '-' || coalesce(cast(updated_at as text), '') as text)) as _airbyte_table_one_hashid
```

## SCDテーブル
#### `_airbyte_unique_key`
指定されたPrimary Keyをハッシュ化した値。以下のクエリが実行されている。
```sql
md5(cast(coalesce(cast("id" as text), '') as text)) as _airbyte_unique_key,
```

#### `_airbyte_unique_key_scd`
**SCDテーブルにおいて一意の値である。**
`_airbyte_unique_key`, `_airbyte_start_at`, `_airbyte_emitted_at` カラムが使われる。
```sql
md5(cast(coalesce(cast(_airbyte_unique_key as text), '') || '-' || coalesce(cast(_airbyte_start_at as text), '') || '-' || coalesce(cast(_airbyte_emitted_at as text), '') as text)) as _airbyte_unique_key_scd
```
#### `_airbyte_start_at`
同期時のcursor filedの値。
例えば `updated_at` がcursor filedなら以下のクエリである。
```sql
updated_at as _airbyte_start_at
```

#### `_airbyte_end_at`
SCDテーブルの追跡しているIDレコードらにおいて、最新ではなくなった場合に値が入る。
```sql
lag(updated_at) over (
    partition by cast("id" as text)
    order by
        updated_at is null asc,
        updated_at desc,
        _airbyte_emitted_at desc
  ) as _airbyte_end_at
```
**特定のIDにおいて変更データがない場合である最初の行や、有効なレコードはNULLになる。**
変更データがある場合は、Cursor Filed(上記例だとupdated_atカラム)で降順に並べ替えた後、lag関数を用いて、一つ前のCursor Filed値を入れる。

#### `_airbyte_active_row `
重複排除されたテーブルに含める必要がある行を指定するために使用される。
**1なら、有効なレコードという意味。**

```sql
case when row_number() over (
    partition by cast("id" as text)
    order by
        updated_at is null asc,
        updated_at desc,
        _airbyte_emitted_at desc
  ) = 1 then 1 else 0 end as _airbyte_active_row,
```



# おわりに
AirbyteはEL(T)と自ら言うほど、極力加工をしない姿勢だが、同期増分の場合だとTransformをせざる得ない印象...


新しい、[Destination V2](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2) はSCDテーブルを廃止しているようだが、過去データも保持したい場合どうすればいいのか疑問が残る。