Skills Leistungsoptimierung
Wenn ein Skill große Dateien verarbeitet oder häufig aufgerufen wird, wirken sich Leistungsprobleme direkt auf die Benutzererfahrung aus.
Dieser Artikel stellt gängige Techniken zur Verbesserung der Verarbeitungsgeschwindigkeit auf Skriptebene vor.
Häufige Ursachen von Leistungsproblemen
| Source | Typische Symptome | Optimierungsrichtung |
|---|---|---|
| Große Dateien auf einmal laden | Das Lesen einer 100MB-CSV dauert mehrere Sekunden | Blockweises Lesen (Chunk) |
| Wiederholtes Ausführen derselben Berechnung | Statistiken bei jedem Aufruf neu berechnen | Ergebnisse zwischenspeichern |
| Einzelthreadige serielle Verarbeitung | Verarbeitung von 1000 Dateien dauert lange | Parallele Verarbeitung |
| Häufige Festplatten-I/O | Dateien zeilenweise in einer Schleife schreiben | Stapelweises Schreiben |
| Laden unnötiger Abhängigkeiten | import dauert mehr als 2 Sekunden | Lazy Import |
Blockweises Lesen großer Dateien
pandas'chunksizeParameter ermöglicht es, große Dateien in mehrere Stapel aufzuteilen und Block für Block zu verarbeiten, wodurch eine übermäßige Speichernutzung auf einmal vermieden wird.
Beispiel
# Dateipfad: scripts/chunked_reader.py
import pandas as pd
import json
import sys
def process_large_csv(file_path: str, chunk_size: int = 10000) -> dict:
"""
Große CSV-Dateien in Blöcken lesen und verarbeiten
Parameter:
file_path: CSV-Dateipfad
chunk_size: Anzahl der Zeilen pro Block, Standard 10000
Rückgabe:
Zusammengeführte statistische Ergebnisse
"""
total_rows = 0
total_sum = 0.0
chunk_count = 0
# Stück für Stück lesen, ohne die gesamte Datei auf einmal in den Speicher zu laden
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
chunk_count += 1
total_rows += len(chunk)
# Statistiken für jeden Chunk berechnen, die Ergebnisse akkumulieren
if "score" in chunk.columns:
total_sum += chunk["score"].sum()
# Fortschritt in Echtzeit melden
print(f" Batch {chunk_count} verarbeitet, {total_rows} Zeilen akkumuliert", flush=True)
avg = total_sum / total_rows if total_rows > 0 else 0
return {
"status": "success",
"total_rows": total_rows,
"chunks": chunk_count,
"score_avg": round(avg, 2)
}
if __name__ == "__main__":
file_path = sys.argv[1] if len(sys.argv) > 1 else ""
result = process_large_csv(file_path)
print(json.dumps(result, ensure_ascii=False, indent=2))
import pandas as pd
import json
import sys
def process_large_csv(file_path: str, chunk_size: int = 10000) -> dict:
"""
Große CSV-Dateien in Blöcken lesen und verarbeiten
Parameter:
file_path: CSV-Dateipfad
chunk_size: Anzahl der Zeilen pro Block, Standard 10000
Rückgabe:
Zusammengeführte statistische Ergebnisse
"""
total_rows = 0
total_sum = 0.0
chunk_count = 0
# Stück für Stück lesen, ohne die gesamte Datei auf einmal in den Speicher zu laden
for chunk in pd.read_csv(file_path, chunksize=chunk_size):
chunk_count += 1
total_rows += len(chunk)
# Statistiken für jeden Chunk berechnen, die Ergebnisse akkumulieren
if "score" in chunk.columns:
total_sum += chunk["score"].sum()
# Fortschritt in Echtzeit melden
print(f" Batch {chunk_count} verarbeitet, {total_rows} Zeilen akkumuliert", flush=True)
avg = total_sum / total_rows if total_rows > 0 else 0
return {
"status": "success",
"total_rows": total_rows,
"chunks": chunk_count,
"score_avg": round(avg, 2)
}
if __name__ == "__main__":
file_path = sys.argv[1] if len(sys.argv) > 1 else ""
result = process_large_csv(file_path)
print(json.dumps(result, ensure_ascii=False, indent=2))
已处理第 1 批，累计 10000 行
已处理第 2 批，累计 20000 行
已处理第 3 批，累计 28456 行
{
"status": "success",
"total_rows": 28456,
"chunks": 3,
"score_avg": 82.37
}
Ergebnis-Caching
Für Vorgänge, bei denen dieselbe Eingabe dasselbe Ergebnis liefert, speichere das Ergebnis in einer lokalen Datei zwischen, um wiederholte Berechnungen zu vermeiden.
Beispiel
# Dateipfad: scripts/file_cache.py
import hashlib
import json
import os
import time
CACHE_DIR = "/home/claude/.skill_cache"
os.makedirs(CACHE_DIR, exist_ok=True)
def _file_fingerprint(file_path: str) -> str:
"""
Erzeuge einen Fingerabdruck basierend auf Dateipfad, Größe und Änderungszeit
Viel schneller als MD5 von Dateiinhalten, geeignet für große Dateien
"""
stat = os.stat(file_path)
raw = f"{file_path}|{stat.st_size}|{stat.st_mtime}"
return hashlib.md5(raw.encode()).hexdigest()
def get_cached(file_path: str, operation: str):
"""Cache abrufen, None zurückgeben, falls nicht vorhanden"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
if os.path.exists(cache_file):
with open(cache_file) as f:
return json.load(f)
return None
def set_cached(file_path: str, operation: str, result: dict):
"""Cache schreiben"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
with open(cache_file, "w") as f:
json.dump(result, f, ensure_ascii=False)
# Verwendungsbeispiel
def get_stats(file_path: str) -> dict:
# Zuerst Cache prüfen
cached = get_cached(file_path, "stats")
if cached:
print("Cache-Treffer, wiederholte Berechnung wird übersprungen")
return cached
# Cache-Miss, Berechnung ausführen
import pandas as pd
t0 = time.time()
df = pd.read_csv(file_path)
result = {
"rows": len(df),
"cols": len(df.columns),
"elapsed": round(time.time() - t0, 3)
}
# Cache für die nächste Verwendung schreiben
set_cached(file_path, "stats", result)
return result
import hashlib
import json
import os
import time
CACHE_DIR = "/home/claude/.skill_cache"
os.makedirs(CACHE_DIR, exist_ok=True)
def _file_fingerprint(file_path: str) -> str:
"""
Erzeuge einen Fingerabdruck basierend auf Dateipfad, Größe und Änderungszeit
Viel schneller als MD5 von Dateiinhalten, geeignet für große Dateien
"""
stat = os.stat(file_path)
raw = f"{file_path}|{stat.st_size}|{stat.st_mtime}"
return hashlib.md5(raw.encode()).hexdigest()
def get_cached(file_path: str, operation: str):
"""Cache abrufen, None zurückgeben, falls nicht vorhanden"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
if os.path.exists(cache_file):
with open(cache_file) as f:
return json.load(f)
return None
def set_cached(file_path: str, operation: str, result: dict):
"""Cache schreiben"""
key = _file_fingerprint(file_path) + "_" + operation
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
with open(cache_file, "w") as f:
json.dump(result, f, ensure_ascii=False)
# Verwendungsbeispiel
def get_stats(file_path: str) -> dict:
# Zuerst Cache prüfen
cached = get_cached(file_path, "stats")
if cached:
print("Cache-Treffer, wiederholte Berechnung wird übersprungen")
return cached
# Cache-Miss, Berechnung ausführen
import pandas as pd
t0 = time.time()
df = pd.read_csv(file_path)
result = {
"rows": len(df),
"cols": len(df.columns),
"elapsed": round(time.time() - t0, 3)
}
# Cache für die nächste Verwendung schreiben
set_cached(file_path, "stats", result)
return result
Parallele Verarbeitung mehrerer Dateien
Wenn mehrere Dateien verarbeitet werden müssen, verwenden Sieconcurrent.futuresfür die parallele Ausführung, was die Geschwindigkeit um ein Vielfaches verbessern kann.
Beispiel
# Dateipfad: scripts/parallel_process.py
import concurrent.futures
import os
import json
import sys
def process_single_file(file_path: str) -> dict:
"""Verarbeitet eine einzelne Datei (wird parallel aufgerufen)"""
try:
size = os.path.getsize(file_path)
# Tatsächliche Verarbeitungslogik simulieren
return {"file": os.path.basename(file_path),
"size_kb": size // 1024, "status": "ok"}
except Exception as e:
return {"file": file_path, "status": "error", "message": str(e)}
def process_files_parallel(file_paths: list, max_workers: int = 4) -> Liste:
"""
Parallele Verarbeitung mehrerer Dateien
Parameter:
file_paths: Liste der Dateipfade
max_workers: maximale Parallelität, Standard 4 (um zu vermeiden, dass zu viele Prozesse um Ressourcen konkurrieren)
"""
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Alle Aufgaben übermitteln
future_map = {
executor.submit(process_single_file, fp): fp
for fp in file_paths
}
# Ergebnisse in der Reihenfolge des Abschlusses sammeln
for future in concurrent.futures.as_completed(future_map):
result = future.result()
results.append(result)
print(f" Abgeschlossen: {result['file']}")
return results
if __name__ == "__main__":
# Simulieren: alle CSV-Dateien im Upload-Verzeichnis verarbeiten
upload_dir = "/mnt/user-data/uploads"
csv_files = [
os.path.join(upload_dir, f)
for f in os.listdir(upload_dir)
if f.endswith(".csv")
]
if not csv_files:
print("Keine CSV-Dateien gefunden")
sys.exit(0)
print(f"Starte parallele Verarbeitung von {len(csv_files)} Dateien...")
results = process_files_parallel(csv_files)
print(json.dumps(results, ensure_ascii=False, indent=2))
import concurrent.futures
import os
import json
import sys
def process_single_file(file_path: str) -> dict:
"""Verarbeitet eine einzelne Datei (wird parallel aufgerufen)"""
try:
size = os.path.getsize(file_path)
# Tatsächliche Verarbeitungslogik simulieren
return {"file": os.path.basename(file_path),
"size_kb": size // 1024, "status": "ok"}
except Exception as e:
return {"file": file_path, "status": "error", "message": str(e)}
def process_files_parallel(file_paths: list, max_workers: int = 4) -> Liste:
"""
Parallele Verarbeitung mehrerer Dateien
Parameter:
file_paths: Liste der Dateipfade
max_workers: maximale Parallelität, Standard 4 (um zu vermeiden, dass zu viele Prozesse um Ressourcen konkurrieren)
"""
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
# Alle Aufgaben übermitteln
future_map = {
executor.submit(process_single_file, fp): fp
for fp in file_paths
}
# Ergebnisse in der Reihenfolge des Abschlusses sammeln
for future in concurrent.futures.as_completed(future_map):
result = future.result()
results.append(result)
print(f" Abgeschlossen: {result['file']}")
return results
if __name__ == "__main__":
# Simulieren: alle CSV-Dateien im Upload-Verzeichnis verarbeiten
upload_dir = "/mnt/user-data/uploads"
csv_files = [
os.path.join(upload_dir, f)
for f in os.listdir(upload_dir)
if f.endswith(".csv")
]
if not csv_files:
print("Keine CSV-Dateien gefunden")
sys.exit(0)
print(f"Starte parallele Verarbeitung von {len(csv_files)} Dateien...")
results = process_files_parallel(csv_files)
print(json.dumps(results, ensure_ascii=False, indent=2))
开始并行处理 3 个文件...
完成：example_jan.csv
完成：example_mar.csv
完成：example_feb.csv
[{"file": "example_jan.csv", "size_kb": 128, "status": "ok"}, ...]
Checkliste zur Leistungsoptimierung
| Prüfpunkt | Vor der Optimierung | Nach der Optimierung |
|---|---|---|
| Große Dateien lesen | pd.read_csv(file) | pd.read_csv(file, chunksize=10000) |
| Wiederholt Statistiken für dieselbe Datei berechnen | Jedes Mal neu berechnen | Dateifingerabdruck verwenden, um Ergebnisse zwischenzuspeichern |
| Mehrere Dateien verarbeiten | for-Schleife, seriell | ThreadPoolExecutor, parallel |
| Dateien Zeile für Zeile schreiben | for row: f.write(row) | Stapelweise sammeln und auf einmal schreiben |
| Unnötige Importe | Alle Importe am Anfang des Skripts | Import bei Bedarf (Lazy Import) |
Andere ErweiterungenVor der Optimierung messen; nicht nach Intuition optimieren. Verwende den
time.perf_counter()oder@timeitDekorator, um den echten Engpass zu identifizieren, und optimiere dann entsprechend, wobei unnötige Codekomplexität durch vorzeitige Optimierung vermieden wird.