1. はじめに — なぜ SQLite なのか
「データベースを入れる」と聞くと、PostgreSQL や MySQL のサーバー構築を思い浮かべる方が多いかもしれません。しかし製造業の現場 PC では、サーバーが要らない SQLite で十分なケースが圧倒的多数です。
- サーバー不要: 1 つの
.dbファイルだけで動く - 標準同梱: Python に標準で
sqlite3モジュールが入っている。追加インストールなし - 性能: 単独 PC での集計・小〜中規模ログ管理なら PostgreSQL に引けを取らない
- 移植性:
.dbファイルをコピーすればそのまま別 PC で開ける（後述の WAL モード運用中は、直近データが-walファイル側に残るため、コピーはアプリ停止後に行うか-wal/-shmも一緒にコピーする） - 信頼性: ACID トランザクション（書き込みが中途半端な状態で残らない仕組み）が標準。Excel と違って「壊れる」リスクが極小
仕様や機能の詳細は SQLite 公式サイト を参照してください。製造業の現場で 1 億行を超えるような規模であれば PostgreSQL / TimescaleDB を検討しますが、それ以下なら SQLite で十分です。
2. 最小実装
1 つ注意: 本記事では説明のために、章が進むごとに measurements テーブルの定義を段階的に変えていきます。各章のコードを試すときは、既存の manufacturing.db（と -wal / -shm）を削除して新しい DB で実行してください。前の章のテーブルが残っていると「table already exists」や「NOT NULL constraint failed」で失敗します（テーブルを作らず読むだけの 5 章・7 章は、直前の章で作った DB をそのまま使ってください）。
import sqlite3
conn = sqlite3.connect("manufacturing.db")
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS measurements (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
machine TEXT NOT NULL,
value REAL NOT NULL
)
""")
cur.execute(
"INSERT INTO measurements (ts, machine, value) VALUES (?, ?, ?)",
("2026-07-03T10:00:00", "M-01", 25.3),
)
conn.commit()
conn.close()これだけでデータベースが完成します。Excel と違って、ファイル破損リスクが極めて小さく、「複数のスクリプトから同時に追記」しても整合性が保たれます。
3. スキーマ設計の基本
3.1 テーブル分離
製造データはおおまかに「マスタ（変わりにくい情報）」と「トランザクション（時系列で増える情報）」に分かれます。
-- マスタ: 設備
CREATE TABLE machines (
machine_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
line TEXT NOT NULL,
installed_at TEXT
);
-- マスタ: 製品
CREATE TABLE products (
product_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
spec_min REAL,
spec_max REAL
);
-- トランザクション: 測定値
CREATE TABLE measurements (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL, -- ISO 8601 ('YYYY-MM-DDTHH:MM:SS')
machine_id TEXT NOT NULL,
product_id TEXT NOT NULL,
lot TEXT NOT NULL,
value REAL NOT NULL,
FOREIGN KEY (machine_id) REFERENCES machines(machine_id),
FOREIGN KEY (product_id) REFERENCES products(product_id)
);
CREATE INDEX idx_measurements_ts ON measurements(ts);
CREATE INDEX idx_measurements_lot ON measurements(lot);
CREATE INDEX idx_measurements_machine_ts ON measurements(machine_id, ts);2 つ補足があります。第一に、SQLite は FOREIGN KEY を宣言しても既定では強制しません。効かせるには接続のたびに conn.execute("PRAGMA foreign_keys=ON") を実行します。第二に、AUTOINCREMENT は SQLite では通常不要です（INTEGER PRIMARY KEY だけで自動連番になり、AUTOINCREMENT は「削除済み ID の再利用を絶対に許さない」場合にだけ意味を持つ、少し重いオプションです。本記事のコードでは他 DB との読み比べのしやすさを優先して付けたままにしています）。
3.2 SQLite の型の特徴
- 動的型付け: SQLite は宣言した型と実際の値の型が違っても入る。「うっかり文字列が数値カラムに入る」ことが起きうるので、書き込み側で気をつける
- 日時型がない:
TEXT（ISO 8601 文字列）またはINTEGER（UNIX タイムスタンプ）で表現。ISO 8601 文字列が後段の処理で扱いやすい - BOOLEAN がない:
INTEGERの 0 / 1 で代用
3.3 インデックスの設計
「よく WHERE で絞る列」「よく JOIN で使う列」「ソート対象の列」にインデックスを張ります。製造データなら ts（時刻）、machine_id、lot あたりが鉄板です。
4. Python からの効率的な書き込み
4.1 NG: 1 件ずつ commit
# これは遅い: 1 件ごとにディスク同期が走る
for record in records:
conn.execute("INSERT INTO measurements ...", record)
conn.commit()4.2 OK: バッチで commit
with conn: # with conn: が暗黙の commit / rollback を担う
conn.executemany(
"INSERT INTO measurements (ts, machine_id, product_id, lot, value) VALUES (?, ?, ?, ?, ?)",
records,
)executemany + 1 回の commit で、1000 件単位の書き込みが瞬時に終わります。1 件ずつ commit する場合と比べて 100 倍以上速くなることもあります——commit のたびにディスクへの同期書き込みが走るため、特に HDD では「1 秒あたりのトランザクション回数」が物理的なボトルネックになるからです。
4.3 WAL モードに切り替える
conn.execute("PRAGMA journal_mode=WAL")WAL（Write-Ahead Logging）モードに切り替えると、書き込み中も別プロセスから読み取りができるようになります。監視アプリ（書き込み）とダッシュボード（読み取り）が同じ DB を共有するときに必須です。詳細な動作仕様は SQLite 公式の WAL モード解説 を参照してください。
⚠️ ここで現場最大の落とし穴を 1 つ。WAL モードは「同一 PC 内」のプロセス間共有専用です。共有メモリ（-shm ファイル）を使う仕組み上、公式ドキュメントも「全プロセスが同じホスト上にあること」を前提とし、ネットワークファイルシステム上では動作しないと明記しています。「NAS や共有フォルダに .db を置いて複数 PC からアクセス」は、WAL 以前にファイルロックの実装差で DB 破損のリスクがある構成なので、やらないでください（複数 PC で共有したくなったら 10 章へ）。
5. SELECT と pandas の連携
ここから pandas を使います（4 章までと違い要インストール: pip install pandas。プロキシ・閉域網での入れ方は 学習ロードマップ STEP 2 参照）。
import pandas as pd
import sqlite3
conn = sqlite3.connect("manufacturing.db")
# クエリ結果を直接 DataFrame に
df = pd.read_sql_query(
"""
SELECT ts, machine_id, value
FROM measurements
WHERE ts >= '2026-07-01'
AND machine_id = 'M-01'
ORDER BY ts
""",
conn,
parse_dates=["ts"],
)
# あとは pandas 流に集計（周期は小文字の "1h"。大文字 "1H" は現行 pandas ではエラー）
hourly = df.set_index("ts")["value"].resample("1h").mean()
print(hourly.head())周期指定の "1h" は昔の記事だと "1H"（大文字）で書かれていますが、pandas 3.x では大文字表記は ValueError になります（筆者環境 pandas 3.0 で実測）。pd.read_sql_query は SQLite の結果を直接 DataFrame に変換します。「DB から最低限のデータを引いて、pandas で集計する」という流れが、Excel で頑張っていた集計の置き換えとして極めて効果的です。
6. ロガーアプリのテンプレート
機器から定周期でデータを取得し、SQLite に書き込む典型的なロガーアプリの骨格です。
# logger.py
import logging
import sqlite3
import time
from pathlib import Path
DB_PATH = Path("manufacturing.db")
INTERVAL = 1.0
logger = logging.getLogger(__name__)
def init_db(conn: sqlite3.Connection) -> None:
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL") # WAL なら NORMAL で十分高速
conn.execute("""
CREATE TABLE IF NOT EXISTS measurements (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
machine_id TEXT NOT NULL,
value REAL NOT NULL
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_measurements_ts ON measurements(ts)")
conn.commit()
def main() -> None:
conn = sqlite3.connect(DB_PATH)
init_db(conn)
buffer: list[tuple] = []
BATCH = 60 # 1 分ぶんを 1 トランザクションで書く
try:
while True:
t0 = time.monotonic()
value = read_from_device()
buffer.append((
time.strftime("%Y-%m-%dT%H:%M:%S"),
"M-01",
value,
))
if len(buffer) >= BATCH:
with conn:
conn.executemany(
"INSERT INTO measurements (ts, machine_id, value) VALUES (?, ?, ?)",
buffer,
)
logger.info("wrote %d rows", len(buffer))
buffer.clear()
elapsed = time.monotonic() - t0
time.sleep(max(0.0, INTERVAL - elapsed))
except KeyboardInterrupt:
pass # Ctrl+C は正常な停止として扱う（finally の flush は実行される）
finally:
# 終了時に残りバッファを flush
if buffer:
with conn:
conn.executemany(
"INSERT INTO measurements (ts, machine_id, value) VALUES (?, ?, ?)",
buffer,
)
conn.close()
def read_from_device() -> float:
# 仮実装: 実機通信に置き換え
import random
return 25.0 + random.uniform(-1.0, 1.0)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
main()「1 件ずつではなく 60 件単位で書く」「終了時にバッファを flush する」「WAL モードで読み込みを邪魔しない」——この 3 点だけで、運用に耐えるロガーになります。
トレードオフも正直に書いておきます。BATCH = 60 は「停電・ブレーカー断で最大 60 秒ぶんのデータが消える」設計です（finally は Ctrl+C には効きますが電源断には無力）。また synchronous=NORMAL は速度と引き換えに、電源断時に直近コミットが失われる可能性を許容する設定です。ロット紐付けなどトレーサビリティ用途なら、BATCH を小さくする・synchronous=FULL にするなど、消えて困らない範囲に調整してください（ソフトで縮める代わりに、現場 PC に UPS＝無停電電源装置を挟んでハードで守る手もあります）。
7. データの肥大化対策
7.1 古いデータの集約
「1 秒ごとの生データを永続保持」は現実的ではありません。古いデータは集約（1 分平均、1 時間平均）して別テーブルに保存し、生データは N ヶ月で削除する設計にします。
-- 1 分平均テーブル
CREATE TABLE measurements_1min (
ts TEXT NOT NULL,
machine_id TEXT NOT NULL,
avg_value REAL NOT NULL,
min_value REAL NOT NULL,
max_value REAL NOT NULL,
n INTEGER NOT NULL,
PRIMARY KEY (ts, machine_id)
);
-- 集約クエリ（毎分実行）
-- 閾値は strftime で「T 区切り + localtime」に揃えるのが急所。
-- 1) datetime('now') は UTC を返す（8.4 参照）
-- 2) datetime() の出力は「2026-07-04 09:10:12」とスペース区切りで、
-- T 区切りの ts と辞書順比較するとズレる（筆者環境で実測）
INSERT OR REPLACE INTO measurements_1min
SELECT
strftime('%Y-%m-%dT%H:%M:00', ts) AS ts,
machine_id,
AVG(value), MIN(value), MAX(value), COUNT(*)
FROM measurements
WHERE ts >= strftime('%Y-%m-%dT%H:%M:%S', 'now', 'localtime', '-1 minute')
GROUP BY 1, 2;この集約は、6 章のロガーの書き込みループに「1 分に 1 回実行」の分岐を足すか、次のような数行のスクリプトにして Windows タスクスケジューラで定期実行します。
# aggregate.py — タスクスケジューラから毎分実行する例
import sqlite3
AGG_SQL = """（上の INSERT OR REPLACE 文をここに貼る）"""
with sqlite3.connect("manufacturing.db") as conn:
conn.execute(AGG_SQL)7.2 古いデータの削除
DELETE FROM measurements
WHERE ts < strftime('%Y-%m-%dT%H:%M:%S', 'now', 'localtime', '-90 days');
VACUUM; -- ファイルサイズを実際に縮めるVACUUM は時間がかかる重い処理で、実行中は一時的に DB サイズの約 2 倍のディスク空きも必要です。ロガーを一時停止できるタイミングで実行してください。24 時間稼働で「静かな時間」がない現場では、日々は DELETE だけ行い、VACUUM は設備停止日（年末・大型連休の保全日など）にまとめて実行する運用が現実的です。
8. 落とし穴と対処
8.1 「database is locked」エラー
WAL モードでない場合、書き込み中の別プロセスからの読み込みでこのエラーが出ます。WAL モードに切り替えるのが第一の対処です。ただし注意——WAL が解決するのは「読み取り vs 書き込み」の競合で、書き込み同士の競合は WAL でも起きます（書き手は常に 1 プロセスだけ）。書き込みプロセスが複数あるなら、Python の sqlite3.connect(DB_PATH, timeout=10) のように待ち時間を延ばすか（既定 5 秒）、そもそも書き込みは 1 プロセスに集約する設計にします。
8.2 マルチスレッドからのアクセス
sqlite3.Connection は同じスレッドからしか使えないのがデフォルトです。複数スレッドから書きたい場合は check_same_thread=False を指定するか、スレッドごとに connection を持ちます。1 connection を複数スレッドで共有しないのが基本です。
8.3 タイムスタンプの形式統一
ISO 8601 の 'YYYY-MM-DDTHH:MM:SS' 形式に統一しておくと、datetime() や strftime() で楽に扱えます。2026/5/3 10:00 のような自由形式は後で必ず後悔します。
8.4 SQLite の 'now' は UTC
datetime('now') や strftime('%Y-%m-%d', 'now') は協定世界時（UTC）を返します。日本の現場では 9 時間ズレるため、ローカル時刻で記録した ts と比較すると「直近 1 分のはずが全期間ヒット」「90 日境界がズレる」といった不可解な挙動になります（筆者環境で実測: datetime('now') = 16:11 のとき datetime('now','localtime') = 翌 01:11）。日本時間で比較するときは必ず 'localtime' 修飾子を付けてください。
8.5 大量 SELECT のメモリ消費
3 年分の生データを丸ごと read_sql_query すると、メモリを食いつぶします。chunksize= 引数で分割読み込みするか、SQL 側で集約してから読みます。
9. SQLite を Excel ユーザーに見せる
SQLite を導入したものの、現場の担当者は今まで通り「Excel で開いて確認したい」というニーズが残ります。次のいずれかで対応できます。
- Streamlit ダッシュボード（Python だけでブラウザ画面が作れるライブラリ）を立てて、ブラウザで見てもらう。構築手順は Streamlit ダッシュボードの記事 を参照
- 定期エクスポート: 日次バッチで
SELECT結果を Excel / CSV に書き出して共有フォルダに置く - DB Browser for SQLite（無料 GUI ツール）を担当者の PC に入れる。ZIP（ポータブル）版なら管理者権限なしで使える
10. PostgreSQL に進むタイミング
SQLite で運用を続ける限界は、おおむね以下の症状で見えてきます。
- 2 台目の PC からネットワーク越しにアクセスしたくなった（SQLite は共有フォルダ上での運用が推奨されず、WAL も使えない。これが最も早く来る移行サイン）
- 書き込みプロセスが複数になり、書き込み競合が頻発
- 1 つの DB ファイルが 10GB を超え、扱いが重くなる
- 分析用に SQLite にない機能（マテリアライズドビュー、豊富な権限管理など）が必要（なおウィンドウ関数自体は SQLite 3.25 以降で使えます）
これらの症状が出たら、PostgreSQL（または TimescaleDB / DuckDB）への移行を検討します。データモデルが SQL 標準に近いので、SQLite で書いたコードは比較的素直に移植できます。
11. おわりに
SQLite は「軽量で・標準同梱で・サーバー不要で・信頼できる」という点で、製造業の現場 PC で扱う計測データの保管先として最適です。Excel から SQLite への移行は、外注に出すような大改修ではなく、Python が書ける現場エンジニアが数日で実装できる規模の改善です。
「測定データ_v3_最新_本物.xlsx」のようなファイルが共有フォルダに転がっている現場に、まずは SQLite を 1 つ導入してみる——その小さな一歩が、現場のデータ運用を大きく変えます。