Pythonでメモリ使用量を監視する方法と活用10選

import psutil
import os

# 現在のプロセスのPIDを取得
pid = os.getpid()

# プロセスオブジェクトを取得
process = psutil.Process(pid)

# メモリ使用量を取得（バイト単位）
memory_info = process.memory_info()

print(f"使用メモリ: {memory_info.rss / 1024 / 1024:.2f} MB")

from memory_profiler import profile

@profile
def memory_hungry_function():
    # メモリを多く消費する処理
    big_list = [i for i in range(1000000)]
    del big_list

if __name__ == '__main__':
    memory_hungry_function()

Line #    Mem usage    Increment   Line Contents
================================================
     4     15.7 MiB     15.7 MiB   @profile
     5                             def memory_hungry_function():
     6                                 # メモリを多く消費する処理
     7     54.0 MiB     38.3 MiB       big_list = [i for i in range(1000000)]
     8     15.7 MiB    -38.3 MiB       del big_list

import sys

# 異なるデータ型のメモリ使用量を比較
int_var = 42
float_var = 3.14
str_var = "Hello, World!"
list_var = [1, 2, 3, 4, 5]
dict_var = {"a": 1, "b": 2, "c": 3}

print(f"整数: {sys.getsizeof(int_var)} バイト")
print(f"浮動小数点数: {sys.getsizeof(float_var)} バイト")
print(f"文字列: {sys.getsizeof(str_var)} バイト")
print(f"リスト: {sys.getsizeof(list_var)} バイト")
print(f"辞書: {sys.getsizeof(dict_var)} バイト")

整数: 28 バイト
浮動小数点数: 24 バイト
文字列: 53 バイト
リスト: 120 バイト
辞書: 232 バイト

import objgraph
import random

class Node:
    def __init__(self, name):
        self.name = name
        self.connections = []

# ノードを作成
nodes = [Node(f"Node{i}") for i in range(10)]

# ランダムに接続
for node in nodes:
    connections = random.sample(nodes, 3)
    node.connections.extend(connections)

# 最も参照されているオブジェクトを表示
objgraph.show_most_common_types()

# 特定のオブジェクトの参照関係を可視化
objgraph.show_backrefs([nodes[0]], filename='node_refs.png')

function                   397
dict                       395
tuple                      267
list                       231
type                       187
Node                        10

import tracemalloc

def allocate_memory():
    return [0] * 1000000

# トラッキング開始
tracemalloc.start()

# メモリを割り当てる前のスナップショット
snapshot1 = tracemalloc.take_snapshot()

# メモリを割り当てる
data = allocate_memory()

# メモリを割り当てた後のスナップショット
snapshot2 = tracemalloc.take_snapshot()

# スナップショットの差分を表示
top_stats = snapshot2.compare_to(snapshot1, 'lineno')

print("[ Top 10 differences ]")
for stat in top_stats[:10]:
    print(stat)

[ Top 10 differences ]
<filename>:<lineno>: size=7.63 MiB (+7.63 MiB), count=2 (+2), average=3.81 MiB

import gc

class CircularReference:
    def __init__(self):
        self.ref = None

# 循環参照を作成
a = CircularReference()
b = CircularReference()
a.ref = b
b.ref = a

# 参照カウントを表示
print(f"aの参照カウント: {gc.get_referrers(a)}")
print(f"bの参照カウント: {gc.get_referrers(b)}")

# ガベージコレクションを実行
collected = gc.collect()

print(f"回収されたオブジェクト数: {collected}")

# 循環参照が解消されたか確認
print(f"aの参照カウント: {gc.get_referrers(a)}")
print(f"bの参照カウント: {gc.get_referrers(b)}")

aの参照カウント: [{'ref': <__main__.CircularReference object at 0x...>}, <__main__.CircularReference object at 0x...>]
bの参照カウント: [{'ref': <__main__.CircularReference object at 0x...>}, <__main__.CircularReference object at 0x...>]
回収されたオブジェクト数: 4
aの参照カウント: [{'ref': <__main__.CircularReference object at 0x...>}]
bの参照カウント: [{'ref': <__main__.CircularReference object at 0x...>}]

import psutil
import time
import logging

# ログの設定
logging.basicConfig(filename='memory_log.txt', level=logging.INFO, 
                    format='%(asctime)s - %(message)s')

def log_memory_usage():
    process = psutil.Process()
    memory_info = process.memory_info()
    logging.info(f"メモリ使用量: {memory_info.rss / 1024 / 1024:.2f} MB")

# 10秒ごとにメモリ使用量をログに記録
for _ in range(6):  # 1分間記録
    log_memory_usage()
    time.sleep(10)

2024-04-01 12:00:00 - メモリ使用量: 14.25 MB
2024-04-01 12:00:10 - メモリ使用量: 14.27 MB
2024-04-01 12:00:20 - メモリ使用量: 14.28 MB
2024-04-01 12:00:30 - メモリ使用量: 14.28 MB
2024-04-01 12:00:40 - メモリ使用量: 14.29 MB
2024-04-01 12:00:50 - メモリ使用量: 14.30 MB

import psutil
import time
import matplotlib.pyplot as plt

def get_memory_usage():
    process = psutil.Process()
    return process.memory_info().rss / 1024 / 1024  # MB単位

# データ収集
time_points = []
memory_usage = []

for i in range(60):  # 1分間データを収集
    time_points.append(i)
    memory_usage.append(get_memory_usage())
    time.sleep(1)

# グラフの描画
plt.figure(figsize=(10, 6))
plt.plot(time_points, memory_usage)
plt.title('メモリ使用量の推移')
plt.xlabel('時間 (秒)')
plt.ylabel('メモリ使用量 (MB)')
plt.grid(True)
plt.savefig('memory_usage_graph.png')
plt.close()

import psutil
import time
import smtplib
from email.mime.text import MIMEText

def send_alert_email(subject, body):
    sender = "your_email@example.com"
    recipient = "alert_recipient@example.com"
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = sender
    msg['To'] = recipient

    with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login(sender, "your_password")
        server.send_message(msg)

def check_memory_usage(threshold_mb):
    process = psutil.Process()
    memory_usage = process.memory_info().rss / 1024 / 1024  # MB単位
    if memory_usage > threshold_mb:
        subject = "メモリ使用量警告"
        body = f"メモリ使用量が {threshold_mb}MB を超えました。現在の使用量: {memory_usage:.2f}MB"
        send_alert_email(subject, body)
        print(f"警告: メモリ使用量が {threshold_mb}MB を超えました。")

# 5秒ごとにメモリ使用量をチェック
threshold = 100  # 100MB
while True:
    check_memory_usage(threshold)
    time.sleep(5)

def number_generator(n):
    for i in range(n):
        yield i

# ジェネレータを使用
gen = number_generator(1000000)

# 最初の10個の数字を表示
for _ in range(10):
    print(next(gen))

0
1
2
3
4
5
6
7
8
9

import weakref

class Node:
    def __init__(self, value):
        self.value = value
        self.parent = None
        self.children = []

    def add_child(self, child):
        self.children.append(child)
        child.parent = weakref.ref(self)

# ノードを作成
root = Node("Root")
child1 = Node("Child 1")
child2 = Node("Child 2")

# 親子関係を設定
root.add_child(child1)
root.add_child(child2)

# 親ノードを参照
print(child1.parent().value)  # 出力: Root

def process_large_file(filename):
    with open(filename, 'r') as file:
        for line in file:
            # 各行を処理
            processed_line = line.strip().upper()
            yield processed_line

# 大きなファイルを1行ずつ処理
for processed_line in process_large_file('large_file.txt'):
    print(processed_line)

from memory_profiler import profile

@profile
def leaky_function():
    big_list = []
    for _ in range(1000000):
        big_list.append(object())
    # big_listを返すことで、関数の外でも参照が保持される
    return big_list

# メモリリークを引き起こす
result = leaky_function()

# メモリを大量に使用する方法
def memory_intensive():
    return [i ** 2 for i in range(10000000)]

# メモリ効率の良い方法
def memory_efficient():
    for i in range(10000000):
        yield i ** 2

# メモリ効率の良い方法を使用
for item in memory_efficient():
    # 各アイテムを処理
    print(item)

class ObjectPool:
    def __init__(self, create_func, max_size=10):
        self.create_func = create_func
        self.max_size = max_size
        self.pool = []

    def get(self):
        if self.pool:
            return self.pool.pop()
        return self.create_func()

    def return_obj(self, obj):
        if len(self.pool) < self.max_size:
            self.pool.append(obj)

# 使用例
def create_expensive_object():
    # 生成コストの高いオブジェクトを作成
    return [0] * 1000000

pool = ObjectPool(create_expensive_object, max_size=5)

# オブジェクトを取得して使用
obj = pool.get()
# オブジェクトを使用した後、プールに返却
pool.return_obj(obj)

import tensorflow as tf
import psutil
import os

def monitor_memory_usage():
    process = psutil.Process(os.getpid())
    mem_usage = process.memory_info().rss / 1024 ** 2  # MBに変換
    return mem_usage

# モデルの定義
model = tf.keras.Sequential([
    tf.keras.layers.Dense(1024, activation='relu', input_shape=(784,)),
    tf.keras.layers.Dense(512, activation='relu'),
    tf.keras.layers.Dense(10, activation='softmax')
])

# メモリ使用量の監視
print(f"モデル作成前のメモリ使用量: {monitor_memory_usage():.2f} MB")

# 大きな入力データの生成
large_input = tf.random.normal((10000, 784))

# 推論の実行
predictions = model(large_input)

print(f"推論後のメモリ使用量: {monitor_memory_usage():.2f} MB")

# メモリの解放
tf.keras.backend.clear_session()

print(f"メモリ解放後のメモリ使用量: {monitor_memory_usage():.2f} MB")

モデル作成前のメモリ使用量: 95.23 MB
推論後のメモリ使用量: 753.46 MB
メモリ解放後のメモリ使用量: 98.12 MB

from flask import Flask, jsonify
import psutil
import os

app = Flask(__name__)

def get_memory_usage():
    process = psutil.Process(os.getpid())
    mem_info = process.memory_info()
    return mem_info.rss / 1024 ** 2  # MBに変換

@app.route('/memory')
def memory_usage():
    return jsonify({
        'memory_usage_mb': get_memory_usage()
    })

@app.route('/heavy_task')
def heavy_task():
    # メモリを大量に使用するタスク
    large_list = [i for i in range(1000000)]
    return jsonify({
        'task': 'completed',
        'memory_usage_mb': get_memory_usage()
    })

if __name__ == '__main__':
    app.run(debug=True)

import csv
import psutil
import os

def get_memory_usage():
    process = psutil.Process(os.getpid())
    mem_info = process.memory_info()
    return mem_info.rss / 1024 ** 2  # MBに変換

def process_csv_inefficient(filename):
    data = []
    with open(filename, 'r') as f:
        reader = csv.reader(f)
        for row in reader:
            data.append(row)
    return len(data)

def process_csv_efficient(filename):
    count = 0
    with open(filename, 'r') as f:
        reader = csv.reader(f)
        for row in reader:
            count += 1
    return count

# 大きなCSVファイルを生成
with open('large_file.csv', 'w') as f:
    writer = csv.writer(f)
    for i in range(1000000):
        writer.writerow([f"データ{i}" for _ in range(10)])

print(f"処理前のメモリ使用量: {get_memory_usage():.2f} MB")

# 非効率な方法
result_inefficient = process_csv_inefficient('large_file.csv')
print(f"非効率な処理後のメモリ使用量: {get_memory_usage():.2f} MB")
print(f"処理した行数: {result_inefficient}")

# メモリをクリア
import gc
gc.collect()

print(f"メモリクリア後のメモリ使用量: {get_memory_usage():.2f} MB")

# 効率的な方法
result_efficient = process_csv_efficient('large_file.csv')
print(f"効率的な処理後のメモリ使用量: {get_memory_usage():.2f} MB")
print(f"処理した行数: {result_efficient}")

処理前のメモリ使用量: 15.23 MB
非効率な処理後のメモリ使用量: 1285.67 MB
処理した行数: 1000000
メモリクリア後のメモリ使用量: 15.45 MB
効率的な処理後のメモリ使用量: 15.48 MB
処理した行数: 1000000