DatabricksにおけるBloomフィルターインデックスのノートブックのウォークスルー

こちらのサンプルノートブックをウォークスルーします。

https://qiita.com/taka_yayoi/items/76c6fe647d334d9650b6

翻訳版のノートブックです。

https://github.com/taka-yayoi/public_repo/blob/main/bloom_filter_index/Bloom%E3%83%95%E3%82%A3%E3%83%AB%E3%82%BF%E3%83%BC%E3%82%A4%E3%83%B3%E3%83%87%E3%83%83%E3%82%AF%E3%82%B9%E3%81%AE%E3%83%87%E3%83%A2.sql

# イントロダクション

Bloomフィルターは、偽陽性の確率を用いてセットにキーがあるかどうか、セットにアイテムが _あるかもしれない_ ことを教えてくれます。インデックスで使用すると、Bloomフィルターは他のテクニックでは高速化できない様なフィールドに対して「藁山から針を探す」ようなクエリーの高速化に役立ちます。

このノートブックでは以下のことを行います:

- テーブルをセットアップし、テーブルにBloomフィルターを設定し、テーブルをハッシュ値で埋めます。
- 「藁山から針を探す」クエリーを実行します:
  - インデックスが作成されていないカラム
  - インデックスが作成されているカラム
  - 結果が0になることが予想されるインデックスが作成されたカラム(藁山に針がない場合にどうなるか)
- サンプルデータセットのクリーンアップ

# テーブル、インデックスの作成、データのロード

## Bloomフィルターインデックスの有効化

:::note info
**注意**
デフォルトで以下の設定は有効になっていますが、デモのため明示的に設定しています。
:::

```sql:SQL
SET spark.databricks.io.skipping.bloomFilter.enabled = true;
```

ユーザー間でデータベース名やファイルパスが重複しない様にします。

```py:Python
%python
import re
from pyspark.sql.types import * 

# Username を取得
username_raw = dbutils.notebook.entry_point.getDbutils().notebook().getContext().tags().apply('user')
# Username の英数字以外を除去し、全て小文字化。Username をファイルパスやデータベース名の一部で使用可能にするため。
username = re.sub('[^A-Za-z0-9]+', '', username_raw).lower()

# データベース名
db_name = f"bloom_filter_{username}"

# パス
data_path = f"dbfs:/tmp/{username_raw}/bloom_test"

# Hiveメタストアのデータベースの準備:データベースの作成
spark.sql(f"DROP DATABASE IF EXISTS {db_name} CASCADE")
spark.sql(f"CREATE DATABASE IF NOT EXISTS {db_name}")
# Hiveメタストアのデータベースの選択
spark.sql(f"USE {db_name}")

print("database name: " + db_name)
```

![Screen Shot 2022-10-04 at 10.52.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d798f5f6-ab1a-fb15-1ca1-f6d602a0cdbb.png)


## テーブルの作成

```py:Python
%python
spark.sql(f"""
CREATE OR REPLACE TABLE bloom_test (
  id   BIGINT NOT NULL,
  str1 STRING NOT NULL,
  sha  STRING NOT NULL,
  sha1 STRING NOT NULL,
  sha2_256 STRING NOT NULL,
  row_hash_too_big STRING NOT NULL,
  row_hash STRING NOT NULL
)
USING DELTA
LOCATION '{data_path}'
""")
```

## データを追加する前にインデックスを作成

```sql:SQL
CREATE BLOOMFILTER INDEX
ON TABLE bloom_test
FOR COLUMNS(sha OPTIONS (fpp=0.1, numItems=50000000))
```

## データ生成

```sql:SQL
TRUNCATE TABLE bloom_test;

WITH sample (
  SELECT
    id,
    'windows.exe' as str1,
    monotonically_increasing_id() mono_id,
    hash(id) hash,
    sha (cast(id % 50000000 as string)) sha,
    sha1(cast(id % 50000000 as string)) sha1,
    sha2(cast(id as string), 256)    sha2_256
  from
    RANGE(0, 1000000000, 1, 448)  -- start, end, step, numPartitions
)
INSERT INTO bloom_test 
SELECT id, 
  str1, 
  sha,
  sha1,
  sha2_256,
  sha2(concat_ws('||',id, str1, mono_id, hash, sha, sha1, sha2_256),512) row_hash_too_big,
  sha2(concat_ws('||',id, str1, mono_id, hash, sha, sha1, sha2_256),256) row_hash
FROM sample
-- LIMIT 20000
```

![Screen Shot 2022-10-04 at 10.54.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a3bd8f89-9245-6508-d335-8042508566c0.png)

以下の様なデータがインサートされます。10億レコードのテーブルとなります。
![Screen Shot 2022-10-04 at 10.54.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3fc4df10-f8f2-398f-731d-1c8277d725af.png)

Z-orderingしておきます。

```sql:SQL
SET spark.databricks.delta.optimize.maxFileSize = 1610612736;
OPTIMIZE bloom_test
ZORDER BY id
```

# 物理テーブルとインデックスを確認

```py:Python
%python
display(dbutils.fs.ls(data_path))
```
![Screen Shot 2022-10-04 at 10.56.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f2cf05e1-1373-90cd-48d4-3cfc8789c833.png)

```py:Python
%python
display(dbutils.fs.ls(f"{data_path}/_delta_index"))
```
![Screen Shot 2022-10-04 at 10.56.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4bf7d8ee-9728-257e-67e0-b153f4b7d931.png)

```sql:SQL
DESCRIBE EXTENDED bloom_test
```
![Screen Shot 2022-10-04 at 10.57.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5a682c73-029c-e118-1df3-b3881d063157.png)

# テストクエリーの実行

こちらは8コアのワーカー、最大10台のオートスケール構成のクラスターを使用しています。

## 同じファイルに存在しないであろうハッシュ値の検索

```sql:SQL
SELECT * FROM bloom_test WHERE id in ( 0, 1, 999999998, 999999999)
```
![Screen Shot 2022-10-04 at 10.58.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9f2f9d92-937a-e96f-fd73-d1dc132c0a97.png)

## Bloomフィルターのインデックスが無いカラムに対するクエリー

```sql:SQL
SELECT count(*) FROM bloom_test WHERE sha1 = 'b2f9544427aed7b712b209fffc756c001712b7ca'
```

20秒程度かかっています。
![Screen Shot 2022-10-04 at 11.02.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/acf5d8fb-d840-e22b-c2e5-462e8127d90d.png)

## Bloomフィルターのインデックスがあるカラムに対するクエリー

```sql:SQL
SELECT count(*) FROM bloom_test WHERE sha = 'b6589fc6ab0dc82cf12099d1c2d40ab994e8410c'
```

5秒程度で結果が返ってきます。
![Screen Shot 2022-10-04 at 11.04.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/eacd9428-5907-1d7a-b3e7-9b13fed94e8c.png)

## Bloomフィルターのインデックスに対して存在しないものを検索

ハッシュ値の末尾に`_`を追加しています。

```sql:SQL
SELECT count(*) FROM bloom_test WHERE sha = 'b6589fc6ab0dc82cf12099d1c2d40ab994e8410_'
```

こちらも5秒程度で結果が返ってきます。
![Screen Shot 2022-10-04 at 11.06.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/71cc0f85-3f22-cbb0-87e4-33d33a514a49.png)

このように10億レコードという比較的大きな規模のテーブルに対しても、Bloomフィルターインデックスを設定することで、「藁山から針を探す(needle in a haystack)」ようなクエリーを高速に処理できる様になります。

### Databricks 無料トライアル

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