Databricks Auto Loader  利用時にファイルが取り込まれない事象への対応方法：チェックポイントの初期化を忘れているケース


## 概要

Databricks Auto Loader  利用時にファイルが取り込まれないと相談を受けることがあるのですが、チェックポイントの初期化を忘れているケースがよくあるため事象を説明します。開発時などに同じファイルを複数回取り込むことがあり、チェックポイントの仕様を失念してファイルが取り込まれないと勘違いされる方がいます。下記のようにチェックポイントの初期化を実施することを忘れないでください。

```python
# チェックポイントを初期化
checkpoint_dir = "dbfs:/user/hive/warehouse/auto_loader_test.db/checkpoint/tbl_1"
dbutils.fs.rm(checkpoint_dir, True)
```

Databricks Auto Loader では、checkpoint のディレクトリにて取り込む済みファイルの一覧を管理しており、その一覧によりファイル取り込みの冪等性を担保しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/de04153e-c6df-05ab-1f7f-15bd08d01f4f.png)

引用元：[自動ローダー - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/ingestion/auto-loader/#how-does-auto-loader-track-ingestion-progress)

## 事象の再現方法

### スキーマとテーブルの作成

```sql
%sql
DROP SCHEMA IF EXISTS auto_loader_test CASCADE;
CREATE SCHEMA IF NOT EXISTS auto_loader_test;
CREATE
OR REPLACE TABLE auto_loader_test.tbl_1 (
  n_nationkey  string
  ,n_name       string
  ,n_regionkey  string 
  ,n_comment    string,
  ingest_date date,
  ingest_timestamp TIMESTAMP,
  _rescued_data string,
  _metadata_file_path string
);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8dbd424d-e235-865e-6e34-65275aeaed0b.png)


### ソースファイルを配置

```python
# ソースのディレクトリを初期化後にファイルを配置
src_dir = "dbfs:/user/hive/warehouse/auto_loader_test.db/src_data/nation"
dbutils.fs.rm(src_dir, True)
tgt_path = f"{src_dir}/ingest_date=2020-01-01/ingest_timestamp=2020-01-01 12%3A34%3A56/csv_file_01.csv"
nation_df = spark.table("samples.tpch.nation")
nation_df.write.format("csv").option("header", True).mode("overwrite").save(tgt_path)

# 配置したファイルを確認
src_df = spark.read.format("csv").option("header", True).load(tgt_path)
src_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4a675338-e3a9-1a3f-5258-54e9f117db1e.png)

### `db_auto_loader_by_parameters.py`というファイル名でコードをソースファイルとして import

```python
# Databricks notebook source
import json

from pyspark.sql.functions import expr

# COMMAND ----------

# ウィジェットの初期化
dbutils.widgets.removeAll()

# COMMAND ----------

# パラメータの受け取り
debug_pipe_conf = """{
    "src_dir": "dbfs:/user/hive/warehouse/auto_loader_test.db/src_data/nation",
    "checkpoint_dir": "dbfs:/user/hive/warehouse/auto_loader_test.db/checkpoint/tbl_1",
    "tgt_tbl_name": "hive_metastore.auto_loader_test.tbl_1"
}
"""
pipe_conf_wiget_name = "pipe_conf"
dbutils.widgets.text(pipe_conf_wiget_name, debug_pipe_conf)
pipe_conf_wiget_value = dbutils.widgets.get(pipe_conf_wiget_name)
print(f"`pipe_conf_wiget_value`: {pipe_conf_wiget_value}")

# COMMAND ----------

# 変数定義
pipe_conf = json.loads(pipe_conf_wiget_value)

src_dir = pipe_conf["src_dir"]
checkpoint_dir = pipe_conf["checkpoint_dir"]
tgt_tbl_name = pipe_conf["tgt_tbl_name"]

# COMMAND ----------

# Databricks Auto Loader の実行
src_df = (
    spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", checkpoint_dir)
    .option("cloudFiles.inferColumnTypes", False)
    .option("cloudFiles.partitionColumns", "ingest_date,ingest_timestamp")
    .option("inferSchema", True)
    .load(src_dir)
)

add_cols = {
    "_metadata_file_path": expr("_metadata.file_path"),
    "ingest_date": expr("to_date(ingest_date)"),
    "ingest_timestamp": expr("try_to_timestamp(replace(ingest_timestamp, '%3A', ':'))"),
}

src_df = src_df.withColumns(add_cols)

(
    src_df.writeStream
    .option("mergeSchema", True)
    .option("checkpointLocation", checkpoint_dir)
    .trigger(availableNow=True)
    .toTable(tgt_tbl_name)
)


# COMMAND ----------

# 事後処理
spark.sql(f"VACUUM {tgt_tbl_name}")
spark.sql(f"ANALYZE TABLE {tgt_tbl_name} COMPUTE STATISTICS FOR ALL COLUMNS")

# COMMAND ----------

# 終了処理
retun_value = json.dumps({})

dbutils.notebook.exit(retun_value)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6ff855e7-483c-4b2a-9003-3bf1447aef83.png)

### `db_auto_loader_by_parameters`ノートブックを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5f738cdb-acb6-ce75-f821-f79621bf6165.png)

### テーブルの再作成を実施

```sql
%sql
CREATE
OR REPLACE TABLE auto_loader_test.tbl_1 (
  n_nationkey  string
  ,n_name       string
  ,n_regionkey  string 
  ,n_comment    string,
  ingest_date date,
  ingest_timestamp TIMESTAMP,
  _rescued_data string,
  _metadata_file_path string
);
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/12faeff9-41cf-f76e-e3d9-2226398f25a5.png)

### `db_auto_loader_by_parameters`ノートブックを再実行し、Databricks Auto Loader を実行しているセルの出力結果が変わったことを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/53706309-5fe3-10e1-9e13-39e6a70c74e1.png)

### 書き込み先のテーブルにデータが存在しないことを確認

```python
# 書き込み先のテーブルを表示
spark.table("hive_metastore.auto_loader_test.tbl_1").display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7638f657-cbda-6dfa-c943-cd00b8b246ba.png)


## 事象への対応方法

### チェックポイントのディレクトリを初期化

```python
# チェックポイントを初期化
checkpoint_dir = "dbfs:/user/hive/warehouse/auto_loader_test.db/checkpoint/tbl_1"
dbutils.fs.rm(checkpoint_dir, True)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ed3e51d9-2ad6-91de-5bed-690990a71aec.png)


### `db_auto_loader_by_parameters`ノートブックを実行

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fe668088-5131-b65d-10c9-99787fde0b3c.png)

### 書き込み先のテーブルにデータが存在することを確認

```python
# 書き込み先のテーブルを表示
spark.table("hive_metastore.auto_loader_test.tbl_1").display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5b8966c6-797e-a26a-16ed-9bc506389d81.png)