OpenSearchとOpenSearch Dashboardsによるダッシュボード作成（ローカルDocker）

{
  "salesData": [
    {
      "date": "2024-11-01",
      "products": [
        {
          "productId": "P001",
          "productName": "コーヒーメーカー",
          "unitsSold": 75
        },
        {
          "productId": "P002",
          "productName": "ノートパソコン",
          "unitsSold": 42
        },
        {
          "productId": "P004",
          "productName": "スマートウォッチ",
          "unitsSold": 65
        },
        // ... 他の商品データ（unitsSold > 0 のみ）
      ]
    },
    {
      "date": "2024-11-02",
      "products": [
        {
          "productId": "P001",
          "productName": "コーヒーメーカー",
          "unitsSold": 58
        },
        {
          "productId": "P003",
          "productName": "Bluetoothスピーカー",
          "unitsSold": 77
        },
        // ... 他の商品データ（unitsSold > 0 のみ）
      ]
    },
    // ... 11月30日までのデータ
  ]
}

import json
import os
import random
from datetime import datetime, timedelta

# 商品リストの定義
products = [
    {"productId": "P001", "productName": "コーヒーメーカー"},
    {"productId": "P002", "productName": "ノートパソコン"},
    {"productId": "P003", "productName": "Bluetoothスピーカー"},
    {"productId": "P004", "productName": "スマートウォッチ"},
    {"productId": "P005", "productName": "エアフライヤー"},
    {"productId": "P006", "productName": "タブレット端末"},
    {"productId": "P007", "productName": "LEDデスクランプ"},
    {"productId": "P008", "productName": "ワイヤレスイヤホン"},
    {"productId": "P009", "productName": "デジタルカメラ"},
    {"productId": "P010", "productName": "電動歯ブラシ"}
]

# 売上データを格納するリスト
sales_data = []

# 開始日と終了日の設定
start_date = datetime(2024, 11, 1)
end_date = datetime(2024, 11, 30)

# 各日についてデータを生成
current_date = start_date
while current_date <= end_date:
    date_str = current_date.strftime("%Y-%m-%d")
    daily_sales = {
        "date": date_str,
        "products": []
    }
    for product in products:
        # 20%の確率で売上数が0になる
        if random.random() < 0.2:
            units_sold = 0
        else:
            # 売上数にランダム性を持たせる（例: 20〜100の間でランダム）
            units_sold = random.randint(20, 100)

        if units_sold > 0:
            product_sales = {
                "productId": product["productId"],
                "productName": product["productName"],
                "unitsSold": units_sold
            }
            daily_sales["products"].append(product_sales)
    sales_data.append(daily_sales)
    current_date += timedelta(days=1)

# JSONデータの構造
output_data = {
    "salesData": sales_data
}

# JSONファイルに出力する場合
os.makedirs("sample_data", exist_ok=True)
with open("sample_data/sales_data_november_2024.json", "w", encoding="utf-8") as f:
    json.dump(output_data, f, ensure_ascii=False, indent=2)

# コンソールに出力する場合
print(json.dumps(output_data, ensure_ascii=False, indent=2))

version: '3'
services:
  opensearch:
    image: opensearchproject/opensearch:2.18.0
    container_name: opensearch
    environment:
      discovery.type: "single-node"
      bootstrap.memory_lock: "true"
      OPENSEARCH_JAVA_OPTS: "-Xms2g -Xmx2g"
      # 初期パスワードを一定強いのにしないとコンテナ起動に失敗する
      OPENSEARCH_INITIAL_ADMIN_PASSWORD: "StrongPass_1234!"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    ports:
      - 9200:9200
    volumes:
      - ./opensearch-data1:/usr/share/opensearch/data # Creates volume called opensearch-data1 and mounts it to the container
    networks:
      - opensearch-net

  opensearch-dashboards:
    image: opensearchproject/opensearch-dashboards:2.18.0
    container_name: opensearch-dashboards
    environment:
      OPENSEARCH_HOSTS: '["https://opensearch:9200"]'
    ports:
      - 5601:5601
    depends_on:
      - opensearch
    networks:
      - opensearch-net

networks:
  opensearch-net:
    driver: bridge

docker compose up -d

docker compose ps

# 以下のようになっていればOK
NAME                    COMMAND                  SERVICE                 STATUS              PORTS
opensearch              "./opensearch-docker…"   opensearch              running             9300/tcp, 9600/tcp, 0.0.0.0:9200->9200/tcp, :::9200->9200/tcp, 9650/tcp
opensearch-dashboards   "./opensearch-dashbo…"   opensearch-dashboards   running             0.0.0.0:5601->5601/tcp, :::5601->5601/tcp

docker compose logs opensearch

    environment:
      OPENSEARCH_HOSTS: '["https://opensearch:9200"]'

import json
import requests

OPENSEARCH_URL = "https://127.0.0.1:9200" # HTTPSで接続
USERNAME = "admin"
PASSWORD = "StrongPass_1234!"
INDEX_NAME = "sales-data"

def create_index():
    # 必要に応じてマッピング定義（unitsSoldを数値、dateをdate型）
    mapping = {
        "settings": {
            "number_of_shards": 1
        },
        "mappings": {
            "properties": {
                "date": {"type": "date"},
                "productId": {"type": "keyword"},
                "productName": {"type": "text"},
                "unitsSold": {"type": "integer"}
            }
        }
    }

    # インデックス作成(存在しない場合)
    res = requests.put(f"{OPENSEARCH_URL}/{INDEX_NAME}", auth=(USERNAME, PASSWORD), json=mapping, verify=False)
    if res.status_code not in [200, 201]:
        print("Index creation response:", res.text)
    else:
        print("Index created successfully!")


def index_data(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)

    bulk_lines = []
    for entry in data["salesData"]:
        date = entry["date"]
        for product in entry["products"]:
            # indexアクション行
            bulk_lines.append(json.dumps({"index": {"_index": INDEX_NAME}}))
            # ドキュメント行
            doc = {
                "date": date,
                "productId": product["productId"],
                "productName": product["productName"],
                "unitsSold": product["unitsSold"]
            }
            bulk_lines.append(json.dumps(doc))

    # 最後に改行が必要
    bulk_body = "\n".join(bulk_lines) + "\n"

    # bulk APIで投入
    bulk_res = requests.post(f"{OPENSEARCH_URL}/_bulk", 
                             auth=(USERNAME, PASSWORD), 
                             headers={"Content-Type": "application/json"}, 
                             data=bulk_body, 
                             verify=False)
    if bulk_res.status_code != 200:
        print("Bulk insert error:", bulk_res.text)
    else:
        response = bulk_res.json()
        print(response)
        print("Data indexed successfully!")

if __name__ == "__main__":
    # インデックス作成（既存なら削除してもよい）
    # requests.delete(f"{OPENSEARCH_URL}/{INDEX_NAME}", auth=(USERNAME, PASSWORD), verify=False)
    create_index()
    index_data("sample_data/sales_data_november_2024.json")

http://127.0.0.1:5601