Delta Lakeのチュートリアル

[Tutorial: Delta Lake \| Databricks on AWS](https://docs.databricks.com/delta/tutorial.html) [2022/11/14時点]の翻訳です。

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

このチュートリアルでは、以下を含むDatabricksにおけるDelta Lakeの一般的なオペレーションを紹介します。

- [テーブルの作成](#テーブルの作成)
- [テーブルのUPSERT](#テーブルのupsert)
- [テーブルからの読み込み](#テーブルの読み込み)
- [テーブル履歴の表示](#テーブル履歴の表示)
- [以前のバージョンのテーブルに対するクエリー](#以前のバージョンのテーブルへのクエリータイムトラベル)
- [テーブルの最適化](#テーブルの最適化)
- [Z-orderインデックスの追加](#カラムによるz-order)
- [参照されないファイルのVacuum](#vacuumによるスナップショットのクリーンアップ)

Databricks[クラスター](https://qiita.com/taka_yayoi/items/c5d99cd77fe4bfcf69f0)にアタッチされた[ノートブック](https://qiita.com/taka_yayoi/items/c306161906d6d34e8bd5)で、本書のサンプル、Python、R、Scala、SQLコードを実行することができます。また、[Databricks SQL](https://docs.databricks.com/sql/index.html)の[SQLウェアハウス](https://qiita.com/taka_yayoi/items/23d7789198c2dcd66381)に関連づけられた[クエリー](https://docs.databricks.com/sql/user/queries/queries.html)で本書のSQLコードを実行することができます。

:::note
**注意**
以下のコードサンプルのいくつかでは、スキーマ(データベース)とテーブル・ビューから構成される2レベルの名前空間記述(`default.people10m`など)を用いています。これらのサンプルを[Unity Catalog](https://qiita.com/taka_yayoi/items/15aede468bdca58ec6a3)で使用するには、2レベルの名前空間をカタログ、スキーマ、テーブル・ビューから構成されるUnity Catalogの3レベル名前空間(`main.default.people10m`など)に置き換えてください。
:::

# テーブルの作成

Databricksで作成されるテーブルはすべてデフォルトでDelta Lakeを使用します。

:::note
**注意**
Databricksランタイム8.0以降では、すべての読み込み、書き込み、テーブル作成のコマンドのデフォルトはDelta Lakeとなります。Databricksランタイム7.3 LTSを使用している場合は、フォーマットを指定するために`delta`キーワードを使用することができます。
:::

```py:Python
# Load the data from its source.
df = spark.read.load("/databricks-datasets/learning-spark-v2/people/people-10m.delta")

# Write the data to a table.
table_name = "people_10m"
df.write.saveAsTable(table_name)
```

```r:R
library(SparkR)
sparkR.session()

# Load the data from its source.
df = read.df(path = "/databricks-datasets/learning-spark-v2/people/people-10m.delta")

# Write the data to a table.
table_name = "people_10m"

saveAsTable(
  df = df,
  tableName = table_name
)
```

```scala:Scala
// Load the data from its source.
val people = spark.read.load("/databricks-datasets/learning-spark-v2/people/people-10m.delta")

// Write the data to a table.
val table_name = "people_10m"

people.write.saveAsTable("people_10m")
```

```sql:SQL
DROP TABLE IF EXISTS people_10m;

CREATE TABLE IF NOT EXISTS people_10m
AS SELECT * FROM delta.`/databricks-datasets/learning-spark-v2/people/people-10m.delta`;
```

上のオペレーションによって、データから推定されたスキーマを用いて新規の[マネージドテーブル](https://qiita.com/taka_yayoi/items/e90be9beeb1a744cee0d#%E3%83%9E%E3%83%8D%E3%83%BC%E3%82%B8%E3%83%89%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%A8%E3%81%AF%E4%BD%95%E3%81%A7%E3%81%99%E3%81%8B)が作成されます。Deltaテーブルを作成する際に使用できるオプションに関しては、[CREATE TABLE](https://docs.databricks.com/sql/language-manual/sql-ref-syntax-ddl-create-table.html)をご覧ください。

マネージドテーブルにおいては、Databricksがデータの格納場所を決定します。格納場所を取得するには以下のように[DESCRIBE DETAIL](https://docs.databricks.com/delta/table-details.html)を使用することができます。

```py:Python
display(spark.sql('DESCRIBE DETAIL people_10m'))
```

```r:R
display(sql("DESCRIBE DETAIL people_10m"))
```

```scala:Scala
display(spark.sql("DESCRIBE DETAIL people_10m"))
```

```sql:SQL
DESCRIBE DETAIL people_10m;
```

データをインサートする前にスキーマを指定してテーブルを作成したいと思うかもしれません。SQLを用いてこれを実現することができます。

```sql:SQL
CREATE TABLE IF NOT EXISTS people10m (
  id INT,
  firstName STRING,
  middleName STRING,
  lastName STRING,
  gender STRING,
  birthDate TIMESTAMP,
  ssn STRING,
  salary INT
)

CREATE OR REPLACE TABLE people10m (
  id INT,
  firstName STRING,
  middleName STRING,
  lastName STRING,
  gender STRING,
  birthDate TIMESTAMP,
  ssn STRING,
  salary INT
)
```

また、テーブルを作成するためにDelta Lakeの`DeltaTableBuilder` APIを使用することもできます。DataFrameWriter APIと比べて、このAPIはカラムのコメント、テーブルプロパティのような追加情報や[ジェネレーテッドカラム](https://docs.databricks.com/delta/generated-columns.html)を容易に指定することができます。

:::note
**プレビュー**
本機能は[パブリックプレビュー](https://docs.databricks.com/release-notes/release-types.html)です。
:::

:::note
**注意**
この機能はDatabricksランタイム8.3以降で使用できます。
:::

```py:Python
# Create table in the metastore
DeltaTable.createIfNotExists(spark) \
  .tableName("default.people10m") \
  .addColumn("id", "INT") \
  .addColumn("firstName", "STRING") \
  .addColumn("middleName", "STRING") \
  .addColumn("lastName", "STRING", comment = "surname") \
  .addColumn("gender", "STRING") \
  .addColumn("birthDate", "TIMESTAMP") \
  .addColumn("ssn", "STRING") \
  .addColumn("salary", "INT") \
  .execute()

# Create or replace table with path and add properties
DeltaTable.createOrReplace(spark) \
  .addColumn("id", "INT") \
  .addColumn("firstName", "STRING") \
  .addColumn("middleName", "STRING") \
  .addColumn("lastName", "STRING", comment = "surname") \
  .addColumn("gender", "STRING") \
  .addColumn("birthDate", "TIMESTAMP") \
  .addColumn("ssn", "STRING") \
  .addColumn("salary", "INT") \
  .property("description", "table with people data") \
  .location("/tmp/delta/people10m") \
  .execute()
```

```scala:Scala
// Create table in the metastore
DeltaTable.createOrReplace(spark)
  .tableName("default.people10m")
  .addColumn("id", "INT")
  .addColumn("firstName", "STRING")
  .addColumn("middleName", "STRING")
  .addColumn(
    DeltaTable.columnBuilder("lastName")
      .dataType("STRING")
      .comment("surname")
      .build())
  .addColumn("lastName", "STRING", comment = "surname")
  .addColumn("gender", "STRING")
  .addColumn("birthDate", "TIMESTAMP")
  .addColumn("ssn", "STRING")
  .addColumn("salary", "INT")
  .execute()

// Create or replace table with path and add properties
DeltaTable.createOrReplace(spark)
  .addColumn("id", "INT")
  .addColumn("firstName", "STRING")
  .addColumn("middleName", "STRING")
  .addColumn(
    DeltaTable.columnBuilder("lastName")
      .dataType("STRING")
      .comment("surname")
      .build())
  .addColumn("lastName", "STRING", comment = "surname")
  .addColumn("gender", "STRING")
  .addColumn("birthDate", "TIMESTAMP")
  .addColumn("ssn", "STRING")
  .addColumn("salary", "INT")
  .property("description", "table with people data")
  .location("/tmp/delta/people10m")
  .execute()
```

# テーブルのUPSERT

既存のDeltaテーブルに対する一連の更新とインサートを管理するために、[MERGE INTO](https://docs.databricks.com/sql/language-manual/delta-merge-into.html)文を使用します。例えば、以下の文はソーステーブルからデータを取り込み、ターゲットテーブルにマージします。両方のテーブル行がマッチする場合、Delta Lakeは指定されたエクスプレッションを用いてデータカラムを更新します。行がマッチしない場合、Delta Lakeは新規の行を追加します。このオペレーションは*upsert*と呼ばれるものです。

```sql:SQL
CREATE OR REPLACE TEMP VIEW people_updates (
  id, firstName, middleName, lastName, gender, birthDate, ssn, salary
) AS VALUES
  (9999998, 'Billy', 'Tommie', 'Luppitt', 'M', '1992-09-17T04:00:00.000+0000', '953-38-9452', 55250),
  (9999999, 'Elias', 'Cyril', 'Leadbetter', 'M', '1984-05-22T04:00:00.000+0000', '906-51-2137', 48500),
  (10000000, 'Joshua', 'Chas', 'Broggio', 'M', '1968-07-22T04:00:00.000+0000', '988-61-6247', 90000),
  (20000001, 'John', '', 'Doe', 'M', '1978-01-14T04:00:00.000+000', '345-67-8901', 55500),
  (20000002, 'Mary', '', 'Smith', 'F', '1982-10-29T01:00:00.000+000', '456-78-9012', 98250),
  (20000003, 'Jane', '', 'Doe', 'F', '1981-06-25T04:00:00.000+000', '567-89-0123', 89900);

MERGE INTO people_10m
USING people_updates
ON people_10m.id = people_updates.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;
```

`*`を指定する場合、ターゲットテーブルのすべてのカラムを更新あるいはインサートします。これは、ターゲットテーブルとソーステーブルが同じカラムを持っていることを前提としており、そうで無い場合にはクエリーは解析エラーで失敗します。

`INSERT`オペレーションを実行する際には、テーブルのすべてのカラムを指定しなくてはなりません(例えば、既存データセットでマッチする行がない場合)。しかし、すべての値を更新する必要はありません。

結果を確認するには、テーブルにクエリーを実行します。

```sql:SQL
SELECT * FROM people_10m WHERE id >= 9999998
```

# テーブルの読み込み

以下のサンプルに示すように、テーブル名あるいはテーブルパスを用いてDeltaテーブルにアクセスします。

```py:Python
people_df = spark.read.table(table_name)

display(people_df)

## or

people_df = spark.read.load(table_path)

display(people_df)
```

```r:R
people_df = tableToDF(table_name)

display(people_df)
```

```scala:Scala
val people_df = spark.read.table(table_name)

display(people_df)

\\ or

val people_df = spark.read.load(table_path)

display(people_df)
```

```sql:SQL
SELECT * FROM people_10m;

SELECT * FROM delta.`<path-to-table`;
```

# テーブルへの書き込み

Delta Lakeはテーブルへのデータの書き込みに標準的な構文を使用します。

既存のDeltaテーブルに新規データを原子性を持って追加するには、以下のサンプルのように`append`モードを使用します。

```sql:SQL
INSERT INTO people10m SELECT * FROM more_people
```

```py:Python
df.write.mode("append").saveAsTable("people10m")
```

```scala:Scala
df.write.mode("append").saveAsTable("people10m")
```

# テーブルの更新

Deltaテーブルで述語にマッチするデータを更新することができます。例えば、`people10m`という名前のテーブル、あるいは`/tmp/delta/people-10m`のパスにあるテーブルにおいて、`gender`カラムの略語を`M`と`F`から`Male`と`Female`に置き換えたい場合、以下を実行します。

```sql:SQL
UPDATE people10m SET gender = 'Female' WHERE gender = 'F';
UPDATE people10m SET gender = 'Male' WHERE gender = 'M';

UPDATE delta.`/tmp/delta/people-10m` SET gender = 'Female' WHERE gender = 'F';
UPDATE delta.`/tmp/delta/people-10m` SET gender = 'Male' WHERE gender = 'M';
```

```py:Python
from delta.tables import *
from pyspark.sql.functions import *

deltaTable = DeltaTable.forPath(spark, '/tmp/delta/people-10m')

# Declare the predicate by using a SQL-formatted string.
deltaTable.update(
  condition = "gender = 'F'",
  set = { "gender": "'Female'" }
)

# Declare the predicate by using Spark SQL functions.
deltaTable.update(
  condition = col('gender') == 'M',
  set = { 'gender': lit('Male') }
)
```

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

val deltaTable = DeltaTable.forPath(spark, "/tmp/delta/people-10m")

// Declare the predicate by using a SQL-formatted string.
deltaTable.updateExpr(
  "gender = 'F'",
  Map("gender" -> "'Female'")

import org.apache.spark.sql.functions._
import spark.implicits._

// Declare the predicate by using Spark SQL functions and implicits.
deltaTable.update(
  col("gender") === "M",
  Map("gender" -> lit("Male")));
```

# テーブルからの削除

Deltaテーブルから術後にマッチするデータを削除することができます。例えば、`people10m`という名前のテーブル、あるいは`/tmp/delta/people-10m`のパスにあるテーブルにおいて、`birthDate`の値が`1995`より小さい値を持つ人々のすべての行を削除するには以下を実行します。

```sql:SQL
DELETE FROM people10m WHERE birthDate < '1955-01-01'

DELETE FROM delta.`/tmp/delta/people-10m` WHERE birthDate < '1955-01-01'
```

```py:Python
from delta.tables import *
from pyspark.sql.functions import *

deltaTable = DeltaTable.forPath(spark, '/tmp/delta/people-10m')

# Declare the predicate by using a SQL-formatted string.
deltaTable.delete("birthDate < '1955-01-01'")

# Declare the predicate by using Spark SQL functions.
deltaTable.delete(col('birthDate') < '1960-01-01')
```

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

val deltaTable = DeltaTable.forPath(spark, "/tmp/delta/people-10m")

// Declare the predicate by using a SQL-formatted string.
deltaTable.delete("birthDate < '1955-01-01'")

import org.apache.spark.sql.functions._
import spark.implicits._

// Declare the predicate by using Spark SQL functions and implicits.
deltaTable.delete(col("birthDate") < "1955-01-01")
```

:::note warn
**重要！**
`delete`は最新バージョンのDeltaテーブルからデータを削除しますが、古いバージョンが明示的にvacuumされるまでは物理的なストレージからは削除されません。[Vacuum](https://qiita.com/taka_yayoi/items/89318c12b58e05c728f6)をご覧ください。
:::

# テーブル履歴の表示

テーブルの履歴を参照するには、テーブルバージョン、オペレーション、ユーザーなどを含む来歴情報を提供する[DESCRIBE HISTORY](https://docs.databricks.com/delta/history.html)文を使用します。

```sql:SQL
DESCRIBE HISTORY people_10m
```

# 以前のバージョンのテーブルへのクエリー(タイムトラベル)

Delta Lakeのタイムトラベルによって、Deltaテーブルの過去のスナップショットに対してクエリーを行うことができます。

テーブルの過去のバージョンにクエリーを行うには、`SELECT`文でバージョンあるいはタイムスタンプを指定します。例えば、上記の履歴からバージョン`0`に対してクエリーを行うには以下を実行します。

```sql:SQL
SELECT * FROM people_10m VERSION AS OF 0
```

あるいは、以下を実行します。

```sql:SQL
SELECT * FROM people_10m TIMESTAMP AS OF '2019-01-29 00:37:58'
```

タイムスタンプに関しては、`"2019-01-01"`や`"2019-01-01'T'00:00:00.000Z"`のような日付とタイムスタンプの文字列のみを指定することができます。

DataFrameReaderのオプションを指定することで、以下のPythonのサンプルのように、特定バージョンのテーブルに固定されたDeltaテーブルからデータフレームを作成できるようになります。

```py:Python
df1 = spark.read.format('delta').option('timestampAsOf', '2019-01-01').table("people_10m")

display(df1)
```

あるいは、

```py:Python
df1 = spark.read.format('delta').option('timestampAsOf', '2019-01-01').table("people_10m")

display(df1)
```

詳細は[Work with Delta Lake table history](https://docs.databricks.com/delta/history.html)をご覧ください。

# テーブルの最適化

テーブルに対して多数の変更を加えると、大量の小規模なファイルが生成される場合があります。クエリーの読み込みスピードを改善するには、小さなファイルを大きなファイルにまとめるために`OPTIMIZE`を使用することができます。

```sql:SQL
OPTIMIZE people_10m
```

# カラムによるZ-order

さらに読み込み性能を改善するために、Z-Orderingによって関連する情報を同じファイルセットに共存させることができます。この局所性は読み込むべきデータの量を劇的に削減するために、Delta Lakeによるデータスキッピングアルゴリズムによって自動的に活用されます。データをZ-Orderするには、`ZORDER BY`句に並び替えるカラムを指定します。`gender`でZ-Orderするには以下を実行します。

```sql:SQL
OPTIMIZE people_10m
ZORDER BY (gender)
```

`OPTIMIZE`を実行する際に指定できるオプションについては、[Compact data files with optimize on Delta Lake](https://docs.databricks.com/delta/optimize.html)をご覧ください。

# VACUUMによるスナップショットのクリーンアップ

Delta Lakeは読み込みにおいてスナップショットのアイソレーションを提供し、これはユーザーやジョブがテーブルにクエリーを行っている際においても`OPTIMIZE`を安全に実行できることを意味します。しかし、最終的には古いスナップショットを削除しなくてはならなくなります。`VACUUM`コマンドを実行することでこれを実現することができます。

```sql:SQL
VACUUM people_10m
```

`VACUUM`の効率的な利用法に関しては、[VACUUMを用いてDeltaテーブルの未使用データを削除する](https://qiita.com/taka_yayoi/items/89318c12b58e05c728f6)をご覧ください。

### Databricks 無料トライアル

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