DatabricksにおけるCOPY INTOを用いたデータのロード

[Load data with COPY INTO \| Databricks on AWS](https://docs.databricks.com/ingestion/copy-into/index.html) [2022/6/9時点]の翻訳です。

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

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

COPY INTOでは[COPY INTOでのデータロードに一時的な認証情報を使う](https://docs.databricks.com/ingestion/copy-into/temporary-credentials.html)ことを含み、いくつかの方法でのセキュアなアクセスをサポートしています。

# 空のDelta Lakeテーブル

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

`COPY INTO`コマンドの実行の際、後ほどスキーマを推定できるように空のプレースホルダーのDeltaテーブルを作成することができます。

```sql:SQL
CREATE TABLE IF NOT EXISTS my_table
[COMMENT <table_description>]
[TBLPROPERTIES (<table_properties>)];

COPY INTO my_table
FROM '/path/to/files'
FILEFORMAT = <format>
FORMAT_OPTIONS ('mergeSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');
```

上のSQL文は冪等性があり、Deltaテーブルに確実に一度だけ(exactly-once)データを取り込むようにスケジュールすることができます。

:::note info
**注意**
空のDeltaテーブルは`COPY INTO`以外では使用できません。`INSERT INTO`や`MERGE INTO`ではスキーマレスのDeltaテーブルにデータを書き込むことをサポートしていません。`COPY INTO`でテーブルデータがインサートされた後は、テーブルがクエリー可能になります。
:::

[COPY INTOのターゲットテーブルの作成](https://qiita.com/taka_yayoi/items/4469a75784ffb2afe96e#copy-into%E3%81%AE%E3%82%BF%E3%83%BC%E3%82%B2%E3%83%83%E3%83%88%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%AE%E4%BD%9C%E6%88%90)をご覧ください。

# サンプル

一般的な利用パターンについては、[DatabricksのCOPY INTOを用いた一般的なデータロードのパターン](https://qiita.com/taka_yayoi/items/4469a75784ffb2afe96e)をご覧ください。

以下の例では、[サンプルデータセット(databricks-datasets)](https://qiita.com/taka_yayoi/items/dcf77d0b007fae774ce5)のサンプルデータをテーブルにロードするために、どのようにDeltaテーブルを作成し、`COPY INTO`を使うのかを示しています。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を実行することもできます。

```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
```

# チュートリアル

- [Bulk load data into a table with COPY INTO in Databricks SQL](https://docs.databricks.com/ingestion/copy-into/tutorial-dbsql.html)
- [Bulk load data into a table with COPY INTO with Spark SQL](https://docs.databricks.com/ingestion/copy-into/tutorial-notebook.html)

# リファレンス

- Databricksランタイム7.x以降: [DatabricksのCOPY INTOコマンド](https://qiita.com/taka_yayoi/items/806763b5e2d212e8cdbc)
- Databricksランタイム5.5LTSおよび6.x: [Copy Into \(Delta Lake on Databricks\)](https://docs.databricks.com/spark/2.x/spark-sql/language-manual/copy-into.html)

### Databricks 無料トライアル

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