Livebook から Amazon DynamoDB の NoSQL データベースを操作する

## はじめに

Livebook　から AWS のサービスを操作するシリーズです

今回は Amazon DynamoDB を操作します

Amazon DynamoDB は NoSQL データベースサービスで、超格安でデータベースを運用できます

フルマネージドでサーバレスなので、環境構築とかは気にせず、使った分だけ料金を払えばOKです

https://aws.amazon.com/jp/dynamodb/

今回はテーブル作成・削除とデータ追加、データ取得を Livebook からやってみます

実装したノートブックはこちら

https://github.com/RyoWakabayashi/elixir-learning/blob/main/livebooks/aws/dynamo.livemd

## 事前作業

AWS のアカンウトと、 DynamoDB の権限を持った IAM ユーザーと、その認証情報（ACCESS_KEY_ID と SECRET_ACCESS_KEY）が必要です

https://docs.aws.amazon.com/ja_jp/IAM/latest/UserGuide/id_users_create.html#id_users_create_console

## 実行環境

Livebook 0.8.0 の Docker　イメージを元にしたコンテナで動かしました

https://hub.docker.com/layers/livebook/livebook/0.8.0/images/sha256-6ef3fdd439fd72af9ae01d58e947664a46a8493f031e179cc0e91c08ddac02e5?context=explore

コンテナ定義はこちらを参照

https://github.com/RyoWakabayashi/elixir-learning

## セットアップ

ex_aws_dynamo を中心に、必要なモジュールをインストールします

また、データ分析のために Explorer もインストールします

```elixir
Mix.install([
  {:ex_aws, "~> 2.0"},
  {:ex_aws_dynamo, "~> 4.2"},
  {:poison, "~> 5.0"},
  {:hackney, "~> 1.18"},
  {:sweet_xml, "~> 0.7"},
  {:explorer, "~> 0.4"},
  {:kino, "~> 0.8"}
])
```

エイリアス等の準備をします

```elixir
alias ExAws.Dynamo
alias Explorer.DataFrame
alias Explorer.Series
require Explorer.DataFrame
```

## 認証

入力エリアを用意し、そこに IAM ユーザーの認証情報を入力します

ACCESS_KEY_ID　と SECRET_ACCESS_KEY は秘密情報なので、値が見えないように `Kino.Input.password` を使います

https://qiita.com/RyoWakabayashi/items/debb1d048613ce505988

```elixir
access_key_id_input = Kino.Input.password("ACCESS_KEY_ID")
```

```elixir
secret_access_key_input = Kino.Input.password("SECRET_ACCESS_KEY")
```

操作対象にする DynamoDB テーブルのリージョンもここで入力しておきましょう

```elixir
region_input = Kino.Input.text("REGION")
```

![スクリーンショット 2022-12-03 0.01.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/9c877571-862a-22bd-528f-32d5cb6c28b5.png)

各認証情報を ExAws に渡すためにまとめておきます

秘密情報が実行結果に現れないよう、セルの最後には `"dummy"` を入れておきましょう

```elixir
auth_config = [
  access_key_id: Kino.Input.read(access_key_id_input),
  secret_access_key: Kino.Input.read(secret_access_key_input),
  region: Kino.Input.read(region_input)
]

"dummy"
```

## テーブル作成

今回は Explorer から化石燃料データセットを取得し、そのデータを DynamoDB に入れてみます

```elixir
df = Explorer.Datasets.fossil_fuels()

df
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 11.09.36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/b070cc32-66bd-4db6-a43a-fd2402a12252.png)

作成するテーブルの名前を定義しておきます

```elixir
table_name = "fossil_fuels"
```

テーブルのキーを定義します

DynamoDB においては RDB と違い、キー項目だけ **ちゃんと** 定義しておけば、あとはどんな項目が来てもデータを追加できます

この **ちゃんと** が難しいわけですが、、、

DynamoDB のキーについては以下の記事を参照してください

https://qiita.com/shibataka000/items/e3f3792201d6fcc397fd

非常にややこしいですが、今回は深いことは考えず、 fossil_fuels データセットは年毎、国毎の化石燃料データなので、 `year` をハッシュキー、 `country` をレンジキーにします

```elixir
key_schema = [
  year: :hash,
  country: :range
]
```

それぞれの型も指定します

```elixir
key_types = [
  year: :number,
  country: :string
]
```

テーブルの性能を指定しておきます

これもかなり分かりにくいですが、オンデマンドキャパシティーモードにしておけば、超大量データを高速に書いたり読んだりしていても、基本的にエラーになりません

その代わり、料金は青天井で、読み書きの回数に応じてどんどん高くなります

どれだけのリクエストが来るか想定できない場合に使います

それに対してプロビジョニング済みキャパシティーモードはある程度時間あたりのリクエスト数が想定できる場合に使用します

読み書きにキャパシティーユニットという数値を指定しておけば、それに応じた時間あたりの料金になります

https://aws.amazon.com/jp/dynamodb/pricing/

そして読み書きそれぞれ 25 キャパシティーユニットまでは無料利用枠が適用されます

https://aws.amazon.com/jp/free/?all-free-tier.sort-by=item.additionalFields.SortRank&all-free-tier.sort-order=asc&awsf.Free%20Tier%20Types=*all&awsf.Free%20Tier%20Categories=*all

小規模データベースだったり、一時的に使うだけのデータベースならほぼ無料で使えてしまいます

というわけで、今回は読み書きそれぞれ 1 のキャパシティーユニットでテーブルを作成します

```elixir
read_capacity = 1
write_capacity = 1
```

`Dynamo.create_table` でテーブルを作成します

```elixir
table_name
|> Dynamo.create_table(key_schema, key_types, read_capacity, write_capacity)
|> ExAws.request(auth_config)
```

![スクリーンショット 2022-12-12 11.23.30.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/4881f96c-6ca3-4a04-b036-a0a0dcadeff6.png)

`Dynamo.list_tables`　で作成したテーブルを確認してみましょう

```elixir
Dynamo.list_tables()
|> ExAws.request(auth_config)
```

![スクリーンショット 2022-12-12 11.24.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/c4ec9325-0e3a-dfe6-5921-2908b2c198bd.png)

## データ追加

まず一つだけデータを追加してみましょう

データフレームから1件データを取得します

```elixir
item =
  df
  |> DataFrame.to_rows()
  |> Enum.at(0)
```

![スクリーンショット 2022-12-12 11.26.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/ab3e25a1-5b00-0dc1-1000-bc90b3e0dbac.png)

`Dynamo.put_item` にそのまま渡せば追加できます

```elixir
table_name
|> Dynamo.put_item(item)
|> ExAws.request(auth_config)
```

![スクリーンショット 2022-12-12 11.27.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/2303d1d9-384a-5183-72f2-466d9baf97bd.png)

では全件追加しましょう、と言いたいところですが、書込キャパシティー 1 で全件追加するとすごく時間がかかるため、 2012 年以降の A で始まる国のデータだけに限定します

対象データを確認します

```elixir
df
|> DataFrame.put(
  :initial,
  Series.transform(df["country"], fn country -> String.first(country) end)
)
|> DataFrame.filter(initial == "A")
|> DataFrame.filter(year >= 2012)
|> DataFrame.select(DataFrame.names(df))
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 11.29.41.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/7a1d68e2-e198-8b34-a138-ff04546bbca9.png)

対象データを逐次追加します

速すぎると書き込みキャパシティーを超えてエラーになってしまうため、 `Process.sleep(500)` で速度を抑えます

```elixir
df
|> DataFrame.put(
  :initial,
  Series.transform(df["country"], fn country -> String.first(country) end)
)
|> DataFrame.filter(initial == "A")
|> DataFrame.filter(year >= 2012)
|> DataFrame.select(DataFrame.names(df))
|> DataFrame.to_rows()
|> Enum.with_index()
|> Enum.map(fn {row, index} ->
  IO.inspect(index)

  Process.sleep(500)

  table_name
  |> Dynamo.put_item(row)
  |> ExAws.request(auth_config)
  |> IO.inspect()
end)
```

![スクリーンショット 2022-12-12 11.39.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/68fb436d-04a8-349c-2d53-46a301f332bb.png)

ゆっくり全件追加できました

## データ取得

DynamoDB のデータ取得には `scan` と `query` と `get_item` があります

`scan`　はキー項目による絞り込みができないときに使います

 DynamoDB 上で全件読み込みが行われるため、読込キャパシティーが無駄に消費されるため推奨されません

全件取得する場合は使っても良いです

```elixir
res =
  table_name
  |> Dynamo.scan()
  |> ExAws.request!(auth_config)
```

![スクリーンショット 2022-12-12 11.42.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/6b3e7601-7b1c-be82-9944-de59edb1a893.png)

先に入れていた1件 + まとめて入れた39件で合わせて40件取得できています

ただし、取得した形式が `"bunker_fuels" => %{"N" => "9"},` というように型と値をキーバリューにした特殊な形式になっているため、整形してからデータフレームに入れます

デーコード用の関数を定義します

```elixir
decode = fn item ->
  item
  |> Enum.reduce(%{}, fn {key, type_value}, merged ->
    type =
      type_value
      |> Map.keys()
      |> Enum.at(0)

    value =
      type_value
      |> Map.values()
      |> Enum.at(0)

    parsed =
      case type do
        "N" ->
          if String.contains?(value, ".") do
            String.to_float(value)
          else
            String.to_integer(value)
          end

        "S" ->
          value
      end

    %{key => parsed}
    |> Map.merge(merged)
  end)
end
```

```elixir
decoded_df =
  res["Items"]
  |> Enum.map(&decode.(&1))
  |> DataFrame.new()

decoded_df
|> DataFrame.select(DataFrame.names(df))
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 11.44.42.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/c07a1e48-2315-8168-52bf-a67aed57293a.png)

`scan` に検索条件や上限件数を指定することもできます

```elixir
decoded_df =
  table_name
  |> Dynamo.scan(
    limit: 3,
    expression_attribute_values: [value: 1_000],
    expression_attribute_names: %{"#name" => "total"},
    filter_expression: "#name > :value"
  )
  |> ExAws.request!(auth_config)
  |> then(&(&1["Items"]))
  |> Enum.map(&decode.(&1))
  |> DataFrame.new()

decoded_df
|> DataFrame.select(DataFrame.names(df))
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 11.55.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/574c61bb-fcf0-4ae3-0ac6-bb2221f098c6.png)

`query` はキー項目によって絞り込みを行います

このとき、 DynamoDB はキーによって絞り込まれたパーティションだけを読み込むため、消費される読込キャパシティーが少なくなります

```elixir
decoded_df =
  table_name
  |> Dynamo.query(
    limit: 5,
    expression_attribute_values: [value: 2013],
    expression_attribute_names: %{"#name" => "year"},
    key_condition_expression: "#name = :value"
  )
  |> ExAws.request!(auth_config)
  |> then(&(&1["Items"]))
  |> Enum.map(&decode.(&1))
  |> DataFrame.new()

decoded_df
|> DataFrame.select(DataFrame.names(df))
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 12.00.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/28e523ac-9357-1ae3-b7a2-3a682f8e4941.png)

`get_item` はキー項目を全て指定して特定のデータ1件だけを取得します

```elixir
decoded_df =
  table_name
  |> Dynamo.get_item(%{year: 2012, country: "AFGHANISTAN"})
  |> ExAws.request!(auth_config)
  |> then(&(&1["Item"]))
  |> decode.()
  |> List.wrap()
  |> DataFrame.new()

decoded_df
|> DataFrame.select(DataFrame.names(df))
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 12.01.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/46c71971-e570-5a4e-31c5-0dfe6f7f2e8a.png)

## データ更新

データ更新には `Dynamo.update_item` を使います

テーブル名、キー項目、更新内容を指定します

`update_expression` の場合は `set` を付けます

```elixir
table_name
|> Dynamo.update_item(
  %{year: 2012, country: "AFGHANISTAN"},
  expression_attribute_values: [cement_value: 6],
  expression_attribute_names: %{"#cement_name" => "cement"},
  update_expression: "set #cement_name = :cement_value"
)
|> ExAws.request!(auth_config)
```

![スクリーンショット 2022-12-12 13.11.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/c3025347-de28-9456-8f89-1ac1b93edc04.png)

```elixir
decoded_df =
  table_name
  |> Dynamo.get_item(%{year: 2012, country: "AFGHANISTAN"})
  |> ExAws.request!(auth_config)
  |> then(&(&1["Item"]))
  |> decode.()
  |> List.wrap()
  |> DataFrame.new()

decoded_df
|> DataFrame.select(DataFrame.names(df))
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 13.11.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/805a0f87-ba29-24c0-d52c-70af2960d36f.png)

## データ削除

削除は　　`Dynamo.delete_item` です

```elixir
table_name
|> Dynamo.delete_item(
  %{year: 2012, country: "AFGHANISTAN"}
)
|> ExAws.request!(auth_config)
```

![スクリーンショット 2022-12-12 13.11.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/c3025347-de28-9456-8f89-1ac1b93edc04.png)

```elixir
decoded_df =
  table_name
  |> Dynamo.scan()
  |> ExAws.request!(auth_config)
  |> then(&(&1["Items"]))
  |> Enum.map(&decode.(&1))
  |> DataFrame.new()

decoded_df
|> DataFrame.select(DataFrame.names(df))
|> Kino.DataTable.new()
```

![スクリーンショット 2022-12-12 13.13.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/d266c94e-305e-da3b-1bf0-e772a4730f0d.png)

## テーブル設定更新

一時的に読込キャパシティーや書込キャパシティーを変更したくなった場合は以下のようにしてテーブル設定を更新します

```elixir
table_name
|> Dynamo.update_table(%{
  provisioned_throughput: %{
    read_capacity_units: 3,
    write_capacity_units: 3,
  }
})
|> ExAws.request(auth_config)
```

![スクリーンショット 2022-12-12 13.28.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/a8074bff-1f78-02b3-4933-b68d58a7bd4d.png)

返ってくる値は更新前のテーブル設定になっています

更新後の（最新の）テーブル設定は `Dynamo.describe_table` で取得します

```elixir
table_name
|> Dynamo.describe_table()
|> ExAws.request(auth_config)
```

![スクリーンショット 2022-12-12 13.30.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/5f37e5fa-cc2b-f478-1d6e-d0dd6027810e.png)

## テーブル削除

テーブルを削除します

```elixir
table_name
|> Dynamo.delete_table()
|> ExAws.request(auth_config)
```

![スクリーンショット 2022-12-12 13.31.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/8b41e6c3-7ecb-92b6-4d27-04370f47a005.png)

```elixir
Dynamo.list_tables()
|> ExAws.request(auth_config)
```

![スクリーンショット 2022-12-12 13.31.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1485835/34edeef7-0fa6-5171-f984-73f23b9cfa2d.png)

## まとめ

わざわざデータ解析のためだけに DynamoDB を使うことはありませんが、何らかのデータ（IoT機器から送られてくるセンサー値など）を DynamoDB に溜めていて、それを解析することはあると思います

DynamoDB から scan もしくは query してきたデータを Livebook 上で解析する、という用途になりそうです