Databricksにおけるテーブルのクローン

[Clone a table on Databricks \| Databricks on AWS](https://docs.databricks.com/optimizations/clone.html) [2022/10/28時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

`clone`コマンドを用いてDatabricksにある既存のDelta Lakeテーブルのコピーを作成することができます。ディープあるいはシャロー(浅い)クローンを作成することができます。

クローンはDatabricksランタイムでデフォルトで有効化されている機能です。

また、ParquetテーブルやIcebergテーブルのクローンをサポートしています。[Incrementally clone Parquet and Iceberg tables to Delta Lake](https://docs.databricks.com/ingestion/clone-parquet.html)をご覧ください。

# クローンのタイプ

- *ディープクローン*は、既存テーブルのメタデータに加えて、ソーステーブルのデータをクローンのターゲットにコピーします。さらに、ソーステーブルに対してDeltaテーブルに書き込みを行うストリームを停止し、クローンのターゲットでコピーした地点から再開できる様に、ストリームのメタデータもクローンされます。
- *シャロークローン*は、クローンターゲットにデータファイルをコピーしないクローンです。テーブルメタデータはソースと同じものになります。これらのクローンは低コストで作成できます。

ディープクローン、シャロークローンに対するすべての変更はクローン自身に反映され、ソーステーブルには影響を与えません。

クローンされるメタデータには、スキーマ、パーティション情報、不変性、NULL可能性が含まれます。ディープクローンにおいてのみ、ストリームと[COPY INTO](https://qiita.com/taka_yayoi/items/806763b5e2d212e8cdbc)のメタデータもクローンされます。クローンされないメタデータは、テーブルの説明文と[ユーザー定義のコミットメタデータ](https://docs.databricks.com/delta/custom-metadata.html)です。

:::note warn
**重要！**
- シャロークローンはソースディレクトリのデータファイルを参照します。ソーステーブルで`vacuum`を実行すると、クライアントは参照しているデータファイルを読み込みでくなくなり、`FileNotFoundException`がスローされます。この場合、`replace`を追加してシャロークローンを実行することでクローンをリペアすることができます。これが頻繁に起きるのであれば、ソーステーブルに依存しないディープクローンを使用することを検討してください。
- ディープローンはクローンしたソースに依存しませんが、データとメタデータをコピーするのでコストが高くなります。
- 指定されたパスにすでにテーブルがあるターゲットへの`replace`を用いたクローンは、当該パスにDeltaログが存在しない場合にDeltaログを作成します。`vacuum`を実行することですべての既存データをクリーンアップすることができます
- 既存のDeltaテーブルが存在する場合、作成される新たなコミットにはソーステーブルからの新規メタデータと新規データが含まれます。新たなコミットはインクリメンタルであり、最後のクローンがテーブルにコミットされた以降の新たな変更のみであることを意味します。
- テーブルのクローンは、`Create Table As Select`、`CTAS`と同じではありません。クローンは、データに加えてソーステーブルのメタデータもコピーします。また、クローンはシンプルな構文であり、パーティション、フォーマット、不変性、NULL可能性はソーステーブルから取得されるのでこれらを指定する必要がありません。
- クローンされたテーブルは、ソーステーブルとは独立した履歴を持ちます。クローンされたテーブルに対するタイムトラベルは、ソーステーブルに対して同じ入力をしたとしても別の挙動をします。
:::

```sql:SQL
CREATE TABLE delta.`/data/target/` CLONE delta.`/data/source/` -- Create a deep clone of /data/source at /data/target

CREATE OR REPLACE TABLE db.target_table CLONE db.source_table -- Replace the target

CREATE TABLE IF NOT EXISTS delta.`/data/target/` CLONE db.source_table -- No-op if the target table exists

CREATE TABLE db.target_table SHALLOW CLONE delta.`/data/source`

CREATE TABLE db.target_table SHALLOW CLONE delta.`/data/source` VERSION AS OF version

CREATE TABLE db.target_table SHALLOW CLONE delta.`/data/source` TIMESTAMP AS OF timestamp_expression -- timestamp can be like “2019-01-01” or like date_sub(current_date(), 1)
```

```py:Python
from delta.tables import *

deltaTable = DeltaTable.forPath(spark, pathToTable)  # path-based tables, or
deltaTable = DeltaTable.forName(spark, tableName)    # Hive metastore-based tables

deltaTable.clone(target, isShallow, replace) # clone the source at latest version

deltaTable.cloneAtVersion(version, target, isShallow, replace) # clone the source at a specific version

# clone the source at a specific timestamp such as timestamp=“2019-01-01”
deltaTable.cloneAtTimestamp(timestamp, target, isShallow, replace)
```

```scala:Scala
import io.delta.tables._

val deltaTable = DeltaTable.forPath(spark, pathToTable)
val deltaTable = DeltaTable.forName(spark, tableName)

deltaTable.clone(target, isShallow, replace) // clone the source at latest version

deltaTable.cloneAtVersion(version, target, isShallow, replace) // clone the source at a specific version

deltaTable.cloneAtTimestamp(timestamp, target, isShallow, replace) // clone the source at a specific timestamp
```

```java:Java
import io.delta.tables.*;

DeltaTable deltaTable = DeltaTable.forPath(spark, pathToTable);
DeltaTable deltaTable = DeltaTable.forName(spark, tableName);

deltaTable.clone(target, isShallow, replace) // clone the source at latest version

deltaTable.cloneAtVersion(version, target, isShallow, replace) // clone the source at a specific version

deltaTable.cloneAtTimestamp(timestamp, target, isShallow, replace) // clone the source at a specific timestamp
```

構文の詳細については、[CREATE TABLE CLONE](https://docs.databricks.com/sql/language-manual/delta-clone.html)をご覧ください。

# クローンのメトリクス

:::note
**注意**
Databricksランタイム8.2以降で利用できます。
:::

`CLONE`がオペレーションを完了すると単一の行で以下のメトリクスをレポートします。

- `source_table_size:` クローンされたソーステーブルのサイズ(バイト)
- `source_num_of_files:` ソーステーブルのファイル数
- `num_removed_files:` テーブルが置換される際、現在のテーブルから削除されたファイルの数
- `num_copied_files:` ソースからコピーされたファイルの数(シャロークローンの場合は0)
- `removed_files_size:` 現在のテーブルから削除されたファイルのサイズ(バイト)
- `copied_files_size:` テーブルにコピーされたファイルのサイズ(バイト)

![](https://docs.databricks.com/_images/clone-metrics.png)

# アクセス権

Databricksのテーブルアクセスコントロールとお使いのクラウドプロバイダーのアクセス権を設定する必要があります。

## テーブルアクセスコントロール

ディープクローン、シャロークローンの両方で以下の権限が必要となります。

- ソーステーブルに対する`SELECT`権限。
- 新規テーブルを作成するために`CLONE`を使う場合、テーブルを作成しようとするデータベースにおける`CREATE`権限。
- テーブルを置換するために`CLONE`を使う場合、テーブルに対する`MODIFY`権限。

## クラウドプロバイダーのアクセス権

ディープローンを作成した場合、ディープクローンを読み込むユーザーはクローンのディレクトリに対する読み込み権限が必要となります。クローンに変更を加えるためには、ユーザーはクローンのディレクトリに対する書き込み権限が必要となります。

シャロークローンを作成した場合、シャロークローンを読み込むユーザーは、シャロークローンのソーステーブルにデータファイルが存在しているので、オリジナルテーブルのファイルとクローンのディレクトリに対する読み取り権限が必要となります。クローンに変更を加えるには、クローンのディレクトリに対する書き込み権限が必要となります。

# データのアーカイブにクローンを使う

タイムトラベルで追えるよりも長い期間データを保持する必要や、ディザスターリカバリーのためにデータを保持する必要があるかもしれません。この様な場合、アーカイブのために特定時点のテーブルお状態を保存するためにディープクローヲン作成することができます。また、でィザスターリカバリーのためにソーステーブルの状態を継続的にアップデートし続けることができる、インクリメンタルなアーカイブも可能です。

```sql:SQL
-- Every month run
CREATE OR REPLACE TABLE delta.`/some/archive/path` CLONE my_prod_table
```

# MLモデルの再現性確保にクローンを使う

機械学習を行う際、MLモデルをトレーニングした際の特定のテーブルバージョンをアーカイブしたいと考えるかもしれません。将来のモデルは、アーカイブしたデータセットを用いてテストすることができます。

```sql:SQL
-- Trained model on version 15 of Delta table
CREATE TABLE delta.`/model/dataset` CLONE entire_dataset VERSION AS OF 15
```

# プロダクションテーブルに対する短期の実験でクローンを使う

テーブルを破壊することなしにプロダクションテーブルに対してワークフローをテストするために、簡単にシャロークローンを作成することができます。これによって、すべてのプロダクションデータを保持しつつも、プロダクションワークロードに影響を与えないクローンテーブル上で任意のワークフローを実行することができます。

```sql:SQL
-- Perform shallow clone
CREATE OR REPLACE TABLE my_test SHALLOW CLONE my_prod_table;

UPDATE my_test WHERE user_id is null SET invalid=true;
-- Run a bunch of validations. Once happy:

-- This should leverage the update information in the clone to prune to only
-- changed files in the clone if possible
MERGE INTO my_prod_table
USING my_test
ON my_test.user_id <=> my_prod_table.user_id
WHEN MATCHED AND my_test.user_id is null THEN UPDATE *;

DROP TABLE my_test;
```

# データ共有にクローンを使う

ある組織の別のビジネスユニットが、最新である必要はないですが同じデータにアクセスしたいと考えるかもしれません。直接ソーステーブルにアクセス権を与えるのではなく、別のビジネスユニットに異なるアクセス権を付与したクローンを提供することができます。クローンのパフォーマンスはシンプルなビューを上回ります。

```sql:SQL
-- Perform deep clone
CREATE OR REPLACE TABLE shared_table CLONE my_prod_table;

-- Grant other users access to the shared table
GRANT SELECT ON shared_table TO `<user-name>@<user-domain>.com`;
```

# テーブルプロパティの上書きにクローンを使う

:::note
**注意**
Databricksランタイム7.5以降で利用できます。
:::

テーブルプロパティの上書きは、特に以下の様なケースで有用です。

- 異なるビジネスユニットにデータを共有する際に、オーナーやユーザー情報でテーブルに注釈をつける。
- Deltaテーブルをアーカイブし、タイムトラベルを可能にする際、以下の様にアーカイブテーブルに対して独立に保持期間を指定することができます。

```sql:SQL
CREATE OR REPLACE TABLE archive.my_table CLONE prod.my_table
TBLPROPERTIES (
delta.logRetentionDuration = '3650 days',
delta.deletedFileRetentionDuration = '3650 days'
)
LOCATION 'xx://archive/my_table'
```

```py:Python
dt = DeltaTable.forName(spark, "prod.my_table")
tblProps = {
"delta.logRetentionDuration": "3650 days",
"delta.deletedFileRetentionDuration": "3650 days"
}
dt.clone('xx://archive/my_table', isShallow=False, replace=True, tblProps)
```

```scala:Scala
val dt = DeltaTable.forName(spark, "prod.my_table")
val tblProps = Map(
"delta.logRetentionDuration" -> "3650 days",
"delta.deletedFileRetentionDuration" -> "3650 days"
)
dt.clone("xx://archive/my_table", isShallow = false, replace = true, properties = tblProps)
```

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)