COPY INTO によりデータロードするプログラム開発時に同一ファイルをソースとする場合の注意事項

## 概要

データエンジニアリングのプログラム開発を行う際にテスト用ファイルをソースとすることがよくありますが、COPY INTO により開発する際には同一ファイル名が取り込まれないことに注意が必要です。Delta Lake 形式のテーブルのディレクトリにて取り込み済みファイル一覧の情報が管理されているようで、ディレクトリの初期化が必要となるようです。この仕様を知らずに開発を行うと、処理が完了しているにも関わらず、データが想定通りに取り込まれないと勘違いしてしまいます。

COPY INTO では、ドキュメントに記載のある通り、ソースとなるファイルを取り込むことを一度だけ実行されることが保証されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4922ad40-ee44-c5b6-dae1-74ba14e58685.png)

引用元：[COPY INTO を使用してデータを読み込む - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/ingestion/copy-into/)

COPY INTO を実行すると、`_delta_log` 配下に `_copy_into_log`というディレクトリが作成され、そこで取り込み済みファイルの一覧を管理しているようです。`sst`という拡張子のファイルがあるため、推測ですが、処理時には RocksDB で管理されていそうです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/16a9fab8-e177-d7f1-a5d6-143ec9a83c9c.png)

COPY INTO にて、取り込み済みファイルでも取り込むためのオプションとして`COPY_OPTIONS`に`force`というオプションがありますが、開発時にのみ `true` にすることは現実的ではないです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0e69714b-a637-8214-2352-97da722893df.png)

引用元：[COPY INTO (Databricks SQL) - Azure Databricks - Databricks SQL | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/sql/language-manual/delta-copy-into#parameters)

下記の方法では`COPY INTO`によりデータが再取り込みされなかったため、ディレクトリの初期化（マネージドテーブルの場合には `DROP TABLE`）が必要になるようです。

- DELETE
- TRUNCATE
- CREATE OR REPLACE
- DROP TABLE（外部テーブルであることが前提）

## 検証
### 事前準備


```python
db_name = '_qiita'
tbl_name = f'{db_name}.region'
tbl_path = f'dbfs:/FileStore/qiita_copy/{tbl_name}'
file_path = "dbfs:/databricks-datasets/tpch/data-001/region/region.tbl"

def copy_into_preparations(
    db_name,
    tbl_name,
    file_path
):
    drop_schem_sql = f'DROP SCHEMA IF EXISTS {db_name} CASCADE'
    spark.sql(drop_schem_sql)
 
    create_schem_sql = f'''
    CREATE SCHEMA _qiita
    LOCATION 'dbfs:/FileStore/qiita_copy'
    '''
    spark.sql(create_schem_sql)
 
    create_tbl_sql = f'''
    CREATE TABLE {tbl_name}
    (
        r_regionkey int,
        r_name string,
        r_comment string
    )
    USING delta
    LOCATION 'dbfs:/FileStore/qiita_copy/{tbl_name}'
    '''
    spark.sql(create_tbl_sql)
 
    copy_sql = f'''
    COPY INTO {tbl_name}
    FROM
     (
        SELECT 
          _c0 AS r_regionkey
          ,_c1 AS r_name
          ,_c2 AS r_comment
          FROM
            '{file_path}'
      )
      FILEFORMAT = CSV  
 
      FORMAT_OPTIONS (
        'mergeSchema' = 'true'
        ,'ignoreCorruptFiles' = 'false'
        ,'delimiter' = '|'
        ,'inferSchema' = 'true'
        ,'header' = 'false'
      )
      COPY_OPTIONS (
        'mergeSchema' = 'true'
        ,'force' = 'false'
      )
    '''
    return spark.sql(copy_sql)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d3fb93ee-29ed-aade-ea86-938054f43ca4.png)


```python
copy_into_preparations(
    db_name,
    tbl_name,
    file_path
)


df = spark.table(tbl_name)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/059df5c5-ea3b-d6d6-7d55-24e571d78521.png)

```python
# 2回目の実行
copy_sql = f'''
COPY INTO {tbl_name}
FROM
 (
    SELECT 
      _c0 AS r_regionkey
      ,_c1 AS r_name
      ,_c2 AS r_comment
      FROM
        '{file_path}'
  )
  FILEFORMAT = CSV  
 
  FORMAT_OPTIONS (
    'mergeSchema' = 'true'
    ,'ignoreCorruptFiles' = 'false'
    ,'delimiter' = '|'
    ,'inferSchema' = 'true'
    ,'header' = 'false'
  )
  COPY_OPTIONS (
    'mergeSchema' = 'true'
    ,'force' = 'false'
  )

# 5 件でかわらないことを確認
df= spark.table(tbl_name)
df.display()
'''
spark.sql(copy_sql)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bcd33db9-1acb-fca2-f691-5eae80bc8185.png)


### ディレクトリを確認

```python
# Delta Lake のディレクトリパスを取得
location = spark.sql(f"desc extended {tbl_name}").filter("col_name='Location'").select('data_type').first()[0]
print(location)

display(dbutils.fs.ls(f'{location}/_delta_log'))
Table
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/319805e6-6223-1798-c7da-513ef84dd2df.png)

```python
display(dbutils.fs.ls(f'{location}/_delta_log/_copy_into_log'))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0b08fcb7-aa48-6c02-408a-82f156edd7dc.png)


```python
display(dbutils.fs.ls(f'{location}/_delta_log/_copy_into_log/SSTs'))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/35d5d2bb-e06b-ae9c-7dc0-548675f7a530.png)


```python
# sst ファイル名を取得
sst_file_name = dbutils.fs.ls(f'{location}/_delta_log/_copy_into_log/SSTs')[0].name
print(dbutils.fs.head(f'{location}/_delta_log/_copy_into_log/SSTs/{sst_file_name}'))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/de91bde2-34d5-ef8e-506e-5b0718dd438e.png)

### ディレクトリを初期化した場合の検証

```python
copy_into_preparations(
    db_name,
    tbl_name,
    file_path
)
spark.table(tbl_name).display()

# Delta Lake のディレクトリパスを取得
location = spark.sql(f"desc extended {tbl_name}").filter("col_name='Location'").select('data_type').first()[0]
dbutils.fs.rm(location, True)

drop_tbl_sql = f'''
DROP TABLE {tbl_name}
'''
spark.sql(drop_tbl_sql)
 
create_or_replace_tbl_sql = f'''
CREATE TABLE {tbl_name}
(
    r_regionkey int,
    r_name string,
    r_comment string
)
USING delta
LOCATION '{tbl_path}'
'''
spark.sql(create_or_replace_tbl_sql)
 
spark.table(tbl_name).display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/757d7993-815e-f24d-b099-eccbb692f86d.png)

```python
copy_sql = f'''
COPY INTO {tbl_name}
FROM
 (
    SELECT 
      _c0 AS r_regionkey
      ,_c1 AS r_name
      ,_c2 AS r_comment
      FROM
        '{file_path}'
  )
  FILEFORMAT = CSV  
 
  FORMAT_OPTIONS (
    'mergeSchema' = 'true'
    ,'ignoreCorruptFiles' = 'false'
    ,'delimiter' = '|'
    ,'inferSchema' = 'true'
    ,'header' = 'false'
  )
  COPY_OPTIONS (
    'mergeSchema' = 'true'
    ,'force' = 'false'
  )
'''
spark.sql(copy_sql)

df = spark.table(tbl_name)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1e88ce41-5b0b-c2ed-cb97-b3c84f0c0a15.png)

### `DELETE`した場合の検証

```python
copy_into_preparations(
    db_name,
    tbl_name,
    file_path
)
spark.table(tbl_name).display()

delete_sql = f'DELETE FROM {tbl_name}'
spark.sql(delete_sql)

copy_sql = f'''
COPY INTO {tbl_name}
FROM
 (
    SELECT 
      _c0 AS r_regionkey
      ,_c1 AS r_name
      ,_c2 AS r_comment
      FROM
        '{file_path}'
  )
  FILEFORMAT = CSV  
 
  FORMAT_OPTIONS (
    'mergeSchema' = 'true'
    ,'ignoreCorruptFiles' = 'false'
    ,'delimiter' = '|'
    ,'inferSchema' = 'true'
    ,'header' = 'false'
  )
  COPY_OPTIONS (
    'mergeSchema' = 'true'
    ,'force' = 'false'
  )
'''
spark.sql(copy_sql)

df = spark.table(tbl_name)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8e2ee4e2-d505-d6cf-ee6d-e667bf772522.png)

### `TRUNCATE`した場合の検証


```python
copy_into_preparations(
    db_name,
    tbl_name,
    file_path
)
spark.table(tbl_name).display()

sql = f'TRUNCATE TABLE {tbl_name}'
spark.sql(sql)

copy_sql = f'''
COPY INTO {tbl_name}
FROM
 (
    SELECT 
      _c0 AS r_regionkey
      ,_c1 AS r_name
      ,_c2 AS r_comment
      FROM
        '{file_path}'
  )
  FILEFORMAT = CSV  
 
  FORMAT_OPTIONS (
    'mergeSchema' = 'true'
    ,'ignoreCorruptFiles' = 'false'
    ,'delimiter' = '|'
    ,'inferSchema' = 'true'
    ,'header' = 'false'
  )
  COPY_OPTIONS (
    'mergeSchema' = 'true'
    ,'force' = 'false'
  )
'''
spark.sql(copy_sql)

df_2 = spark.table(tbl_name)
df_2.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2e9940b4-4850-03c7-c1e4-b80a019a4d4d.png)

### `CREATE OR REPLACE`した場合の検証

```python
copy_into_preparations(
    db_name,
    tbl_name,
    file_path
)
spark.table(tbl_name).display()

create_or_replace_tbl_sql = f'''
CREATE OR REPLACE TABLE {tbl_name}
(
    r_regionkey int,
    r_name string,
    r_comment string
)
USING delta
'''
spark.sql(create_or_replace_tbl_sql)

copy_sql = f'''
COPY INTO {tbl_name}
FROM
 (
    SELECT 
      _c0 AS r_regionkey
      ,_c1 AS r_name
      ,_c2 AS r_comment
      FROM
        '{file_path}'
  )
  FILEFORMAT = CSV  
 
  FORMAT_OPTIONS (
    'mergeSchema' = 'true'
    ,'ignoreCorruptFiles' = 'false'
    ,'delimiter' = '|'
    ,'inferSchema' = 'true'
    ,'header' = 'false'
  )
  COPY_OPTIONS (
    'mergeSchema' = 'true'
    ,'force' = 'false'
  )
'''
spark.sql(copy_sql)

df_2 = spark.table(tbl_name)
df_2.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ce80c440-ca81-b66f-517f-5b46c14e6cc0.png)

### 外部テーブルを`DROP TABLE`した場合の検証

```python
copy_into_preparations(
    db_name,
    tbl_name,
    file_path
)
spark.table(tbl_name).display()

# わかりやすくするために、データを削除
sql = f'TRUNCATE TABLE {tbl_name}'
spark.sql(sql)

drop_tbl_sql = f'''
DROP TABLE {tbl_name}
'''
spark.sql(drop_tbl_sql)

create_or_replace_tbl_sql = f'''
CREATE TABLE {tbl_name}
-- (
--     r_regionkey int,
--     r_name string,
--     r_comment string
-- )
USING delta
LOCATION '{tbl_path}'
'''
spark.sql(create_or_replace_tbl_sql)

copy_sql = f'''
COPY INTO {tbl_name}
FROM
 (
    SELECT 
      _c0 AS r_regionkey
      ,_c1 AS r_name
      ,_c2 AS r_comment
      FROM
        '{file_path}'
  )
  FILEFORMAT = CSV  
 
  FORMAT_OPTIONS (
    'mergeSchema' = 'true'
    ,'ignoreCorruptFiles' = 'false'
    ,'delimiter' = '|'
    ,'inferSchema' = 'true'
    ,'header' = 'false'
  )
  COPY_OPTIONS (
    'mergeSchema' = 'true'
    ,'force' = 'false'
  )
'''
spark.sql(copy_sql)

df_2 = spark.table(tbl_name)
df_2.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9014aa91-7761-f080-f9c8-6aa764d53e46.png)

### `COPY_OPTIONS`の`force`オプションの検証

```python
copy_into_preparations(
    db_name,
    tbl_name,
    file_path
)
spark.table(tbl_name).display()

copy_sql = f'''
COPY INTO {tbl_name}
FROM
 (
    SELECT 
      _c0 AS r_regionkey
      ,_c1 AS r_name
      ,_c2 AS r_comment
      FROM
        '{file_path}'
  )
  FILEFORMAT = CSV  
 
  FORMAT_OPTIONS (
    'mergeSchema' = 'true'
    ,'ignoreCorruptFiles' = 'false'
    ,'delimiter' = '|'
    ,'inferSchema' = 'true'
    ,'header' = 'false'
  )
  COPY_OPTIONS (
    'mergeSchema' = 'true'
    ,'force' = 'true'
  )
'''
spark.sql(copy_sql)

# 10レコードあることを確認
df = spark.table(tbl_name)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/243abfe2-d8f8-1024-6e2a-fc3855e07762.png)

### 本手順で作成したリソースを削除

```python
%sql
DROP SCHEMA IF EXISTS _qiita CASCADE
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8aab4330-a5fd-4e5d-9068-aa5b57cd25cd.png)