E-MapReduce:データ管理

DELETE

データを削除します。サンプルコード：

DELETE FROM delta_table [AS t] [WHERE t.date < '2019-11-11'];

import io.delta.tables.
val deltaTable = DeltaTable.forPath(spark, "/tmp/delta_table")
deltaTable.delete("date < '2019-11-11'")
import org.apache.spark.sql.functions.
import spark.implicits.
deltaTable.delete(col("date") < "2019-11-11")

• サブクエリを持つ WHERE 句はサポートされていません。ただし、サブクエリが SQL ステートメントで表現されたスカラサブクエリである場合は、spark.sql.uncorrelated.scalar.subquery.preexecution.enabled パラメーターを true に設定できます。このようにして、サブクエリを持つ WHERE 句を使用できます。例：

  DELETE FROM delta_table WHERE t.date < (SELECT date FROM ref_table WHERE ....)

• 別のテーブルの行と一致する行を削除する場合は、MERGE ステートメントを使用します。たとえば、MERGE ステートメントを使用して DELETE FROM target WHERE target.col = ref.col ... ステートメントを置き換えます。

UPDATE

データを更新します。サンプルコード：

UPDATE delta_table [AS t] SET t.id = t.id + 1 [WHERE t.date < '2019-11-11'];

import io.delta.tables._

val deltaTable = DeltaTable.forPath(spark, "/tmp/delta_table")

deltaTable.updateExpr(            // SQL 形式の文字列を使用します。
  "name = 'Robet'",
  Map("name" -> "'Robert'")

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

deltaTable.update(                // SQL 関数を使用し、暗黙的な変換を実行します。
  col("name") === "Robet"),
  Map("name" -> lit("Robert"));

• サブクエリを持つ WHERE 句はサポートされていません。ただし、サブクエリが SQL ステートメントで表現されたスカラサブクエリである場合は、spark.sql.uncorrelated.scalar.subquery.preexecution.enabled パラメーターを true に設定できます。このようにして、サブクエリを持つ WHERE 句を使用できます。例：

  UPDATE delta_table SET t.id = t.id + 1 WHERE t.date < (SELECT date FROM ref_table WHERE ....)

• 別のテーブルの行と一致する行を更新する場合は、MERGE ステートメントを使用します。たとえば、MERGE ステートメントを使用して UPDATE target SET target.col = ref.col ... または UPDATE target WHERE target.col = ref.col ... ステートメントを置き換えます。

MERGE

データをマージします。サンプルコード：

MERGE INTO target AS t
USING source AS s
ON t.date = s.date
WHEN MATCHED [AND t.name = 'should_update'] THEN UPDATE SET target.name = source.name
WHEN MATCHED [AND t.name = 'should_delete'] THEN DELETE
WHEN NOT MATCHED [AND s.name = 'should_insert'] THEN INSERT (t.date, t.name, t.id) VALUES (s.date, s.name.s.id)

import io.delta.tables._
import org.apache.spark.sql.functions._

val updatesDF = ...  // updates DataFrame[date, id, name] を定義します

DeltaTable.forPath(spark, "/tmp/delta_table")
  .as("target")
  .merge(updatesDF.as("source"), "target.id = source.id")
  .whenMatched("target.name = 'should_update'")
  .updateExpr(Map("target.name" -> "source.name"))
  .whenMatched("target.name = 'should_delete'")
  .delete()
  .whenNotMatched("source.name = 'shoulde_insert'")
  .insertExpr(
    Map(
      "date" -> "updates.date",
      "eventId" -> "updates.eventId",
      "data" -> "updates.data"))
  .execute()

• UPDATE 句または INSERT 句でアスタリスク (*) をワイルドカードとして使用できます。句が UPDATE SET * または INSERT * の場合、すべてのフィールドが更新または挿入されます。

• サブクエリを持つ ON 条件はサポートされていません。ただし、サブクエリが SQL ステートメントで表現されたスカラサブクエリである場合は、spark.sql.uncorrelated.scalar.subquery.preexecution.enabled パラメーターを true に設定できます。このようにして、サブクエリを持つ ON 条件を使用できます。

ALTER TABLE

既存のテーブルのスキーマとプロパティを変更します。ALTER TABLE ステートメントを実行して、テーブルに対して次の操作を実行できます。

• ADD COLUMN：テーブルに列を追加します。

• RENAME COLUMN：列の名前を変更します。このステートメントを使用するには、列マッピングを有効にする必要があります。

• DROP COLUMN：テーブルから列を削除します。このステートメントを使用するには、列マッピングを有効にする必要があります。

• SET/UNSET TBLPROPERTIES：テーブルの説明やストレージ形式など、テーブルのプロパティを設定します。

• RENAME TO：テーブルの名前を変更します。

次のサンプルコードは、指定した位置に列を追加する方法の例を示しています。

-- delta_tbl テーブルのスキーマは (id IN, name STRING, pt STRING) です。pt はパーティションフィールドです。
-- name フィールドの後、pt フィールドの前に new_col フィールドを追加します。
ALTER TABLE dbName.tableName ADD COLUMN (new_col STRING AFTER name);

DESCRIBE HISTORY

Delta テーブルの操作履歴の詳細を表示します。

CONVERT

Parquet テーブルを Delta テーブルに変換します。

CONVERT ステートメントは、特定のパスにあるすべての Parquet データファイルをトラバースし、現在の Parquet テーブルのスキーマを推測し、Delta テーブルに必要なメタデータ情報を生成します。Parquet テーブルがパーティションテーブルの場合、パーティションフィールドとパーティションフィールドのデータ型を指定する必要があります。

CREATE TABLE tbl_without_partition
USING delta
LOCATION "oss://region/path/to/tbl_without_partition";

OPTIMIZE

set spark.databricks.delta.stats.skipping=true;
set spark.databricks.delta.stats.collect=true;

-- dbName.tableName テーブル全体を最適化します。
OPTIMIZE dbName.tableName;

-- date フィールドの値が 2021-04-01 より前のパーティションを最適化します。
OPTIMIZE dbName.tableName WHERE date < '2021-04-01';

-- date フィールドの値が 2021-04-01 より前のパーティションを最適化し、col2 列と col3 列に Z オーダーを実装します。
OPTIMIZE dbName.tableName WHERE date < '2021-04-01' ZORDER BY (col2, col3);

VACUUM

不要になったデータファイルと、テーブルパスから保持期間よりも長く保存されているデータファイルを削除します。

• 構文

  VACUUM (path=STRING | table=tableIdentifier) (RETAIN number HOURS)? (DRY RUN)?

• 例

  -- データファイルを削除します。
  VACUUM dbName.tableName; 
  -- 24 時間以上保持されているデータファイルを削除します。
  VACUUM dbName.tableName RETAIN 24 HOURS; 
  -- 24 時間以上保持されているデータファイルを表示します。
  VACUUM dbName.tableName RETAIN 24 HOURS DRY RUN; 

  説明

  • VACUUM ステートメントを定期的に実行してデータファイルを削除し、より多くのストレージスペースを解放できます。

  • データファイルを削除する前に、DRY RUN で終わる VACUUM ステートメントを実行して、削除されるデータファイルを表示できます。

SAVEPOINT

Delta テーブルの履歴バージョンを永続的に保持します。

Delta Lake が Delta テーブルでチェックポイント操作を実行するたびに、ログのメタデータファイルが削除されます。delta.checkpointInterval パラメーターは、チェックポイント操作の間隔を指定するために使用されます。delta.logRetentionDuration パラメーターは、ログのメタデータファイルの保持期間を指定するために使用されます。デフォルトでは、メタデータファイルは 30 日間保持されます。VACUUM ステートメントを実行すると、履歴バージョンで不要になったデータファイルも削除されます。SAVEPOINT ステートメントを実行して、データファイルとログのメタデータファイルを永続的に保持できます。タイムトラベル機能とともに SAVEPOINT ステートメントを使用して、履歴バージョンのデータを読み取ることができます。

ROLLBACK

Delta テーブルを特定の履歴バージョンにロールバックします。

必要なデータファイルまたはログのメタデータファイルが失われているために指定された履歴バージョンを復元できない場合は、エラーが報告されます。例：

• Delta テーブルを ID が 0 のバージョンにロールバックします。

  ROLLBACK delta.`/path/to/delta_tbl` VERSION AS OF 0;

• Delta テーブルを特定の時点より前にリリースされた最新バージョンにロールバックします。

  ROLLBACK dbName.tableName TIMESTAMP AS OF "2021-04-01 10:00:00";