DatabricksにおけるDelta Lakeへのデータ取り込み

[Ingest data into Delta Lake \| Databricks on AWS](https://docs.databricks.com/delta/delta-ingest.html) [2022/8/29時点]の翻訳です。

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

DatabricksではDelta Lakeにデータを取り込む方法を多数取り揃えています。

# CSVファイルのアップロード

[Databricks SQLのCreate Table UI](https://qiita.com/taka_yayoi/items/89eda6708b7c2e77d651)を用いてCSVファイルからテーブルをセキュアに作成することができます。

# パートナーインテグレーション

Databricksのパートナーインテグレーションを用いることで、簡単にDatabricksにデータをロードすることができます。これらのインテグレーションによって、様々なソースからDatabricksに対して、ローコード、実装が容易かつスケーラブルなデータ取り込みを実現することができます。詳細は[Databricks integrations](https://docs.databricks.com/integrations/index.html)をご覧ください。

# `COPY INTO` SQLコマンド

`COPY INTO` SQLコマンドを用いることで、ファイルの格納場所からDeltaテーブルにデータをロードすることができます。これは再トライ可能で、冪等性のあるオペレーションです。ソースのロケーションにある処理済みのファイルはスキップされます。

以下のケースでは、[Auto Loader](#auto-loader)の代わりに`COPY INTO`を使用します。

- 数千未満のファイルを含むファイルロケーションからデータをロードしたいケース。
- お使いのデータのスキーマは頻繁に変更されないケース。
- 以前ロードしたファイルのサブセットをロードしようとしているケース。

`COPY INTO` SQLコマンドの概要とデモに関しては、本書後半の[Auto Loader](#auto-loader)と以下のYoutube動画(2分)をご覧ください。
<iframe width="560" height="315" src="https://www.youtube.com/embed/Wte44wRZKDk" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen></iframe>

以下の例では、どのようにDeltaテーブルを作成し、[サンプルデータセット(databricks-datasets)](https://qiita.com/taka_yayoi/items/dcf77d0b007fae774ce5)からテーブルにサンプルデータをロードするのかを説明しています。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 info
**注意**
以下のコードサンプルのいくつかでは、スキーマ(データベース)とテーブル・ビューで構成される2レベルの名前空間(例`default.people10m`)を使用しています。これらのサンプルを[Unity Catalog](https://qiita.com/taka_yayoi/items/15aede468bdca58ec6a3)で実行するには2レベルの名前空間を、カタログ、スキーマ、テーブル・ビューから構成されるUnity Catalogの3レベルの名前空間(例`main.default.people10m`)に置き換えてください。詳細は、[Use three\-level namespace notation with Unity Catalog](https://docs.databricks.com/delta/delta-unity-catalog.html#three-level-namespacing)をご覧ください。
:::

```py:Python
table_name = 'default.loan_risks_upload'
source_data = '/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet'
source_format = 'PARQUET'

spark.sql("DROP TABLE IF EXISTS " + table_name)

spark.sql("CREATE TABLE " + table_name + " (" \
  "loan_id BIGINT, " + \
  "funded_amnt INT, " + \
  "paid_amnt DOUBLE, " + \
  "addr_state STRING)"
)

spark.sql("COPY INTO " + table_name + \
  " FROM '" + source_data + "'" + \
  " FILEFORMAT = " + source_format
)

loan_risks_upload_data = spark.sql("SELECT * FROM " + table_name)

display(loan_risks_upload_data)

'''
Result:
+---------+-------------+-----------+------------+
| loan_id | funded_amnt | paid_amnt | addr_state |
+=========+=============+===========+============+
| 0       | 1000        | 182.22    | CA         |
+---------+-------------+-----------+------------+
| 1       | 1000        | 361.19    | WA         |
+---------+-------------+-----------+------------+
| 2       | 1000        | 176.26    | TX         |
+---------+-------------+-----------+------------+
...
'''
```

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

table_name = "default.loan_risks_upload"
source_data = "/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet"
source_format = "PARQUET"

sql(paste("DROP TABLE IF EXISTS ", table_name, sep = ""))

sql(paste("CREATE TABLE ", table_name, " (",
  "loan_id BIGINT, ",
  "funded_amnt INT, ",
  "paid_amnt DOUBLE, ",
  "addr_state STRING)",
  sep = ""
))

sql(paste("COPY INTO ", table_name,
  " FROM '", source_data, "'",
  " FILEFORMAT = ", source_format,
  sep = ""
))

loan_risks_upload_data = tableToDF(table_name)

display(loan_risks_upload_data)

# Result:
# +---------+-------------+-----------+------------+
# | loan_id | funded_amnt | paid_amnt | addr_state |
# +=========+=============+===========+============+
# | 0       | 1000        | 182.22    | CA         |
# +---------+-------------+-----------+------------+
# | 1       | 1000        | 361.19    | WA         |
# +---------+-------------+-----------+------------+
# | 2       | 1000        | 176.26    | TX         |
# +---------+-------------+-----------+------------+
# ...
```

```scala:Scala
val table_name = "default.loan_risks_upload"
val source_data = "/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet"
val source_format = "PARQUET"

spark.sql("DROP TABLE IF EXISTS " + table_name)

spark.sql("CREATE TABLE " + table_name + " (" +
  "loan_id BIGINT, " +
  "funded_amnt INT, " +
  "paid_amnt DOUBLE, " +
  "addr_state STRING)"
)

spark.sql("COPY INTO " + table_name +
  " FROM '" + source_data + "'" +
  " FILEFORMAT = " + source_format
)

val loan_risks_upload_data = spark.table(table_name)

display(loan_risks_upload_data)

/*
Result:
+---------+-------------+-----------+------------+
| loan_id | funded_amnt | paid_amnt | addr_state |
+=========+=============+===========+============+
| 0       | 1000        | 182.22    | CA         |
+---------+-------------+-----------+------------+
| 1       | 1000        | 361.19    | WA         |
+---------+-------------+-----------+------------+
| 2       | 1000        | 176.26    | TX         |
+---------+-------------+-----------+------------+
...
*/
```

```sql:SQL
DROP TABLE IF EXISTS default.loan_risks_upload;

CREATE TABLE default.loan_risks_upload (
  loan_id BIGINT,
  funded_amnt INT,
  paid_amnt DOUBLE,
  addr_state STRING
);

COPY INTO default.loan_risks_upload
FROM '/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet'
FILEFORMAT = PARQUET;

SELECT * FROM default.loan_risks_upload;

-- Result:
-- +---------+-------------+-----------+------------+
-- | loan_id | funded_amnt | paid_amnt | addr_state |
-- +=========+=============+===========+============+
-- | 0       | 1000        | 182.22    | CA         |
-- +---------+-------------+-----------+------------+
-- | 1       | 1000        | 361.19    | WA         |
-- +---------+-------------+-----------+------------+
-- | 2       | 1000        | 176.26    | TX         |
-- +---------+-------------+-----------+------------+
-- ...
```

クリーンアップするには、テーブルを削除する以下のコードを実行します。

```py:Python
spark.sql("DROP TABLE " + table_name)
```

```r:R
sql(paste("DROP TABLE ", table_name, sep = ""))
```

```scala:Scala
spark.sql("DROP TABLE " + table_name)
```

```sql:SQL
DROP TABLE default.loan_risks_upload
```

より詳細やサンプルについては、以下をご覧ください。

- Databricksランタイム7.x以降: [DatabricksのCOPY INTOコマンド](https://qiita.com/taka_yayoi/items/806763b5e2d212e8cdbc)

# Auto Loader

Auto Loaderは追加のセットアップなしに、クラウドストレージに新規のデータファイルが到着すると、インクリメンタルかつ効率的にファイルを処理します。Auto Loaderは`cloudFiles`という新たな構造化ストリーミングソースを提供します。クラウドファイルストレージの入力ディレクトリパスが指定されると、`cloudFiles`ソースは、データが到着すると新規ファイルを自動で処理します。オプションで既にディレクトリにあるファイルを処理することも可能です。

以下のケースでは、[COPY INTO SQLコマンド](#copy-into-sqlコマンド)の代わりにAuto Loaderを使用します。

- 数百万以上の規模のファイルを含むファイルロケーションからデータをロードしたいケース。Auto Loaderは、`COPY INTO` SQLコマンドよりもさらに効率的にファイルを発見し、複数のバッチにファイル処理を分割することができます。
- データスキーマが頻繁に変更されるケース。Auto Loaderはスキーマ推定とスキーマ進化に対して優れたサポートを提供します。詳細は、[Auto Loaderにおけるスキーマ推定とスキーマ進化の設定](https://qiita.com/taka_yayoi/items/2ccef74b279b1477154b)をご覧下さい。
- 以前アップロードしたファイルのサブセットをロードする計画がないケース。Auto Loaderではファイルのサブセットを再処理することが困難となります。しかし、Auto Loaderのストリームが実行されるのと並行して、ファイルのサブセットを理ロードするために`COPY INTO`コマンドを使用することができます。

Auto Loaderの概要とデモに関しては、本書前半の[COPY INTO SQLコマンド](#copy-into-sqlコマンド)と以下のYoutube動画(2分)をご覧ください。
<iframe width="560" height="315" src="https://www.youtube.com/embed/Wte44wRZKDk" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen></iframe>

Auto Loaderの更なる概要説明とデモに関しては、こちらのYouTube動画(59分)をご覧ください。
<iframe width="560" height="315" src="https://www.youtube.com/embed/8a38Fv9cpd8" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen></iframe>

:::note info
**注意**
以下のコードサンプルのいくつかでは、スキーマ(データベース)とテーブル・ビューで構成される2レベルの名前空間(例`default.people10m`)を使用しています。これらのサンプルを[Unity Catalog](https://qiita.com/taka_yayoi/items/15aede468bdca58ec6a3)で実行するには2レベルの名前空間を、カタログ、スキーマ、テーブル・ビューから構成されるUnity Catalogの3レベルの名前空間(例`main.default.people10m`)に置き換えてください。詳細は、[Use three\-level namespace notation with Unity Catalog](https://docs.databricks.com/delta/delta-unity-catalog.html#three-level-namespacing)をご覧ください。
:::

以下のサンプルコードでは、クラウドストレージに新規データファイルが到着するとどのようにAuto Loaderが検知するのかを説明しています。Databricks[クラスター](https://qiita.com/taka_yayoi/items/c5d99cd77fe4bfcf69f0)にアタッチされた[ノートブック](https://qiita.com/taka_yayoi/items/c306161906d6d34e8bd5)でサンプルコードを実行することができます。

1. 以下のようにファイルアップロードディレクトリを作成します。

    ```py:Python
    user_dir = '<my-name>@<my-organization.com>'
    upload_path = "/FileStore/shared-uploads/" + user_dir + "/population_data_upload"
    
    dbutils.fs.mkdirs(upload_path)
    ```

    ```scala:Scala
    val user_dir = "<my-name>@<my-organization.com>"
    val upload_path = "/FileStore/shared-uploads/" + user_dir + "/population_data_upload"
    
    dbutils.fs.mkdirs(upload_path)
    ```

1. 以下のサンプルCSVファイルを作成し、[DBFSファイルブラウザ](https://docs.databricks.com/dbfs/file-browser.html)を用いてファイルアップロードディレクトリにアップロードします。

    `WA.csv`:
    
    ```
    city,year,population
    Seattle metro,2019,3406000
    Seattle metro,2020,3433000
    ```

    `OR.csv`:
    
    ```
    city,year,population
    Portland metro,2019,2127000
    Portland metro,2020,2151000
    ```

1. Auto Loaderを起動するために以下のコードを実行します。

    ```py:Python
    checkpoint_path = '/tmp/delta/population_data/_checkpoints'
    write_path = '/tmp/delta/population_data'
    
    # Set up the stream to begin reading incoming files from the
    # upload_path location.
    df = spark.readStream.format('cloudFiles') \
      .option('cloudFiles.format', 'csv') \
      .option('header', 'true') \
      .schema('city string, year int, population long') \
      .load(upload_path)
    
    # Start the stream.
    # Use the checkpoint_path location to keep a record of all files that
    # have already been uploaded to the upload_path location.
    # For those that have been uploaded since the last check,
    # write the newly-uploaded files' data to the write_path location.
    df.writeStream.format('delta') \
      .option('checkpointLocation', checkpoint_path) \
      .start(write_path)
    ```

    ```scala:Scala
    val checkpoint_path = "/tmp/delta/population_data/_checkpoints"
    val write_path = "/tmp/delta/population_data"
    
    // Set up the stream to begin reading incoming files from the
    // upload_path location.
    val df = spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .option("header", "true")
      .schema("city string, year int, population long")
      .load(upload_path)
    
    // Start the stream.
    // Use the checkpoint_path location to keep a record of all files that
    // have already been uploaded to the upload_path location.
    // For those that have been uploaded since the last check,
    // write the newly-uploaded files' data to the write_path location.
    df.writeStream.format("delta")
      .option("checkpointLocation", checkpoint_path)
      .start(write_path)
    ```

1. ステップ3のコードが動いている間に、書き込み先のディレクトリにクエリーを行うために以下のコードを実行します。

    ```py:Python
    df_population = spark.read.format('delta').load(write_path)
    
    display(df_population)
    
    '''
    Result:
    +----------------+------+------------+
    | city           | year | population |
    +================+======+============+
    | Seattle metro  | 2019 | 3406000    |
    +----------------+------+------------+
    | Seattle metro  | 2020 | 3433000    |
    +----------------+------+------------+
    | Portland metro | 2019 | 2127000    |
    +----------------+------+------------+
    | Portland metro | 2020 | 2151000    |
    +----------------+------+------------+
    '''
    ```

    ```scala:Scala
    val df_population = spark.read.format("delta").load(write_path)
    
    display(df_population)
    
    /* Result:
    +----------------+------+------------+
    | city           | year | population |
    +================+======+============+
    | Seattle metro  | 2019 | 3406000    |
    +----------------+------+------------+
    | Seattle metro  | 2020 | 3433000    |
    +----------------+------+------------+
    | Portland metro | 2019 | 2127000    |
    +----------------+------+------------+
    | Portland metro | 2020 | 2151000    |
    +----------------+------+------------+
    */
    ```

1. ステップ3のコードが動いている間に、以下の追加のCSVファイルを作成し、[DBFSファイルブラウザ](https://docs.databricks.com/dbfs/file-browser.html)を用いてファイルアップロードディレクトリにアップロードします。

    `ID.csv`:
    
    ```
    city,year,population
    Boise,2019,438000
    Boise,2020,447000
    ```

    `MT.csv`:
    
    ```
    city,year,population
    Helena,2019,81653
    Helena,2020,82590
    ```

    `Misc.csv`:
    
    ```
    city,year,population
    Seattle metro,2021,3461000
    Portland metro,2021,2174000
    Boise,2021,455000
    Helena,2021,81653
    ```

1. ステップ3のコードが動いている間に、書き込み先のディレクトリにクエリーを行うために以下のコードを実行すると、Auto Loaderがアップロードディレクトリで新規データを検知し、書き込みディレクトリに書き込みを行っていることがわかります。

    ```py:Python
    df_population = spark.read.format('delta').load(write_path)
    
    display(df_population)
    
    '''
    Result:
    +----------------+------+------------+
    | city           | year | population |
    +================+======+============+
    | Seattle metro  | 2019 | 3406000    |
    +----------------+------+------------+
    | Seattle metro  | 2020 | 3433000    |
    +----------------+------+------------+
    | Helena         | 2019 | 81653      |
    +----------------+------+------------+
    | Helena         | 2020 | 82590      |
    +----------------+------+------------+
    | Boise          | 2019 | 438000     |
    +----------------+------+------------+
    | Boise          | 2020 | 447000     |
    +----------------+------+------------+
    | Portland metro | 2019 | 2127000    |
    +----------------+------+------------+
    | Portland metro | 2020 | 2151000    |
    +----------------+------+------------+
    | Seattle metro  | 2021 | 3461000    |
    +----------------+------+------------+
    | Portland metro | 2021 | 2174000    |
    +----------------+------+------------+
    | Boise          | 2021 | 455000     |
    +----------------+------+------------+
    | Helena         | 2021 | 81653      |
    +----------------+------+------------+
    '''
    ```

    ```scala:Scala
    val df_population = spark.read.format("delta").load(write_path)
    
    display(df_population)
    
    /* Result
    +----------------+------+------------+
    | city           | year | population |
    +================+======+============+
    | Seattle metro  | 2019 | 3406000    |
    +----------------+------+------------+
    | Seattle metro  | 2020 | 3433000    |
    +----------------+------+------------+
    | Helena         | 2019 | 81653      |
    +----------------+------+------------+
    | Helena         | 2020 | 82590      |
    +----------------+------+------------+
    | Boise          | 2019 | 438000     |
    +----------------+------+------------+
    | Boise          | 2020 | 447000     |
    +----------------+------+------------+
    | Portland metro | 2019 | 2127000    |
    +----------------+------+------------+
    | Portland metro | 2020 | 2151000    |
    +----------------+------+------------+
    | Seattle metro  | 2021 | 3461000    |
    +----------------+------+------------+
    | Portland metro | 2021 | 2174000    |
    +----------------+------+------------+
    | Boise          | 2021 | 455000     |
    +----------------+------+------------+
    | Helena         | 2021 | 81653      |
    +----------------+------+------------+
    */
    ```

1. クリーンアップするには、ステップ3で実行しているコードをキャンセルし、アップロードディレクトリ、チェックポイントディレクトリ、書き込みディレクトリを削除する以下のコードを実行します。

    ```py:Python
    dbutils.fs.rm(write_path, true)
    dbutils.fs.rm(upload_path, true)
    ```

    ```scala:Scala
    dbutils.fs.rm(write_path, true)
    dbutils.fs.rm(upload_path, true)
    ```

### Databricks 無料トライアル

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