「負荷テストや処理時間の差をテストしてみたいな〜」と思った時に、最初にぶち当たる問題がダミーデータをどうするかです。
以前、PostgreSQLの処理時間の差を比較した記事を書いた際は適当な文字列を作成しましたが、今考えるとあまりクールなやり方ではありませんでした。
そこでPythonで簡単にダミーデータを作成することができる、Factory BoyとFakerを使ってのダミーデータの生成について紹介します。
やってみよう
まずは、必要なライブラリをインストールします。
pip install factory_boy faker
Factory boy と faker の関係性についてです。
ダミーデータの生成はfakerが行い、Factory boy がモデルやオブジェクトの自動生成をしてくれます。
まずは、fakerを使ってみます。
faker でダミーデータの生成
from faker import Faker
fake = Faker("ja_JP") # 日本語設定
print("名前:", fake.name]())
出力
名前: 村上 香織
ループで回してみると、毎回異なる名前が出力されます。
また名前以外にもたくさんの項目を使用することができます。
from faker import Faker
fake = Faker("ja_JP") # 日本語設定
print("名前:", fake.name())
print("住所:", fake.address())
print("電話番号:", fake.phone_number())
print("メールアドレス:", fake.email())
print("会社名:", fake.company())
print("職業:", fake.job())
print("日付:", fake.date())
print("テキスト:", fake.text())
print("都道府県:", fake.prefecture())
print("市区町村:", fake.city())
print("郵便番号:", fake.postcode())
出力
名前: 前田 結衣
住所: 福岡県三宅島三宅村港南20丁目26番15号 細竹アーバン486
電話番号: 090-2476-1024
メールアドレス: miki21@example.com
会社名: 木村銀行有限会社
職業: 配管工
日付: 1971-01-22
テキスト: リハビリホイール午前管理する参加する目的擁する。キャビネットサンプルテント数字。中央鉱山差別する尿行進。
ノート通行料金ブレーキスマッシュボトル。スキーム革新中世緩む評議会。持つ数字風景カレッジサワー学生。
評議会パーセントヒール錯覚品質キャビン。サワー拡張省略錯覚持つ高い式。通行料金知覚パイオニア状況。部隊分割叔父腐ったキャビネット。
都道府県: 宮崎県
市区町村: 三鷹市
郵便番号: 339-9502
郵便番号などは指定した言語の国の表記に合わせたものになっています。
一部日本語対応していない項目もあるようですが、ほとんど使えます。
ただ都道府県と市区町村の一致などはなかったりするので、東京都大阪市とかになる可能性もあります。その辺の細かい部分は気にしない方が良いでしょう。
また、都道府県は実在するものですが、郵便番号は桁数が合っているだけで実在しないものもランダムで作成されています。郵便番号を正規化している場合などは、エラーの原因になるので気をつけてください。
Factory Boy を使ったオブジェクト生成
Fakerだけでも多種多彩なダミーデータが作れますが、”まとまったデータ構造を大量に作る”となると、毎回手書きで辞書やクラスを書いていると大変です。
そこで登場するのが Factory Boy です。
Factory Boy は、Fakerで生成した値を組み合わせて、クラスやモデルのインスタンスを自動で構築してくれる仕組みを提供してくれます。
実際にやってみましょう。
import factory
from faker import Faker
fake = Faker("ja_JP")
# ダミー用のUserクラスを定義（SQLAlchemyやDjangoで言うモデルに相当）
class User:
def __init__(self, name, email, address):
self.name = name
self.email = email
self.address = address
# Userクラス用のFactoryを定義
class UserFactory(factory.Factory):
class Meta:
model = User # 対象クラスを指定
name = factory.Faker("name", locale="ja_JP")
email = factory.Faker("email", locale="ja_JP")
address = factory.Faker("address", locale="ja_JP")
# 1件作成
user = UserFactory()
print(user.name, user.email, user.address)
出力
鈴木 稔 lokamoto@example.org 大分県横浜市中区勝どき27丁目5番11号 浅草アーバン501
この書き方だけでは、あまりしっくりこないかもしれません。
実際のプロジェクトでは、SQLAlchemyなどのORMと組み合わせて使うことがほとんどだと思います。
今後大きめのデータを使ってRustで遊びたいなと思っていたりするので、10万件の企業データをダミーとして作成してCSV出力をしてみます。
import csv
import os
from concurrent.futures import ThreadPoolExecutor
from typing import List
from faker import Faker
from tqdm import tqdm
# 日本語のダミーデータ生成器（スレッドローカル）
def get_fake() -> Faker:
"""スレッドローカルなFakerインスタンスを取得"""
if not hasattr(get_fake, "_fake"):
get_fake._fake = Faker("ja_JP")
return get_fake._fake
# ファイル名と件数
output_file: str = "companies_100k.csv"
num_rows: int = 100_000
batch_size: int = 10_000 # バッチサイズ
max_workers: int = min(4, os.cpu_count()) # 並列数
# ヘッダー（業種を除いた6列）
header: List[str] = ["企業名", "郵便番号", "都道府県", "住所", "担当者氏名", "電話番号"]
def generate_batch_data(size: int) -> List[List[str]]:
"""指定サイズのバッチデータを生成（スレッドセーフ）"""
fake = get_fake() # スレッドローカルのFakerインスタンス
batch_data = []
for _ in range(size):
batch_data.append(
[
fake.company(),
fake.postcode(),
fake.prefecture(),
fake.address().replace("\n", ""),
fake.name(),
fake.phone_number(),
]
)
return batch_data
def parallel_csv_generation() -> None:
"""並列処理によるCSV生成（最高速）"""
with open(output_file, mode="w", newline="", encoding="utf-8") as csvfile:
writer = csv.writer(csvfile)
writer.writerow(header)
# バッチサイズリストを準備
batch_sizes: List[int] = []
remaining: int = num_rows
while remaining > 0:
current_size: int = min(batch_size, remaining)
batch_sizes.append(current_size)
remaining -= current_size
# 並列でバッチデータ生成
with ThreadPoolExecutor(max_workers=max_workers) as executor:
with tqdm(total=num_rows, desc="CSV出力中", unit="rows") as pbar:
# バッチデータを並列生成
futures = [
executor.submit(generate_batch_data, size) for size in batch_sizes
]
# 完了順に書き込み（順序保持のため結果を収集）
batch_results: List[List[List[str]]] = []
for future in futures:
batch_data: List[List[str]] = future.result()
batch_results.append(batch_data)
# 順序を保持して書き込み
for batch_data in batch_results:
writer.writerows(batch_data)
pbar.update(len(batch_data))
def optimized_csv_generation() -> None:
"""最適化されたCSV生成関数（シングルスレッド版）"""
fake: Faker = Faker("ja_JP")
with open(output_file, mode="w", newline="", encoding="utf-8") as csvfile:
writer = csv.writer(csvfile)
writer.writerow(header)
# バッチ処理で効率化
total_batches: int = (num_rows + batch_size - 1) // batch_size
with tqdm(total=num_rows, desc="CSV出力中", unit="rows") as pbar:
for batch_num in range(total_batches):
# 最後のバッチのサイズ調整
current_batch_size: int = min(
batch_size, num_rows - batch_num * batch_size
)
# バッチデータ生成
batch_data: List[List[str]] = []
for _ in range(current_batch_size):
batch_data.append(
[
fake.company(),
fake.postcode(),
fake.prefecture(),
fake.address().replace("\n", ""),
fake.name(),
fake.phone_number(),
]
)
# バッチ書き込み
writer.writerows(batch_data)
# 進捗更新
pbar.update(current_batch_size)
if __name__ == "__main__":
print(f"📊 {num_rows:,}件のダミーデータ生成を開始...")
print(f"⚙️ 設定: バッチサイズ={batch_size:,}, 並列数={max_workers}")
# 使用する関数を選択（並列処理版 or シングルスレッド版）
use_parallel: bool = True # 並列処理を使う場合はTrue
if use_parallel:
print("🚀 並列処理モードで実行中...")
parallel_csv_generation()
else:
print("⚡ シングルスレッドモードで実行中...")
optimized_csv_generation()
print(f"\n✅ CSVファイルの生成が完了しました: {output_file}")
# ファイルサイズ確認
try:
file_size_mb: float = os.path.getsize(output_file) / (1024**2)
print(f"📁 ファイルサイズ: {file_size_mb:.2f} MB")
except Exception:
print("📁 ファイルサイズの取得に失敗しました")
無事、ダミーデータを生成できました。
今回のコードは生成AIに助けてもらって、並列処理で高速化する方法を教えてもらいました。
生成AIの作ったコードには、なぜかprintの時に絵文字みたいなのが入るのはなぜなんでしょうか。
とりあえず、ダミーデータの生成が完了しました。
負荷テストとかをするときに、データに困ることがなさそうですね。