Распределенные системы на Python: мощь масштабирования и вычислений

# Пример gRPC сервера
import grpc
from concurrent import futures
import time
import hello_pb2
import hello_pb2_grpc

class Greeter(hello_pb2_grpc.GreeterServicer):
def SayHello(self, request, context):
return hello_pb2.HelloReply(message=f"Hello, {request.name}!")

def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
hello_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
server.add_insecure_port('[::]:50051')
server.start()
try:
while True:
time.sleep(86400)
except KeyboardInterrupt:
server.stop(0)

if __name__ == '__main__':
serve()

# tasks.py
from celery import Celery

app = Celery('tasks', 
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/0')

@app.task
def process_data(data):
# Ресурсоемкая операция обработки данных
result = perform_calculations(data)
return result

# В другом файле или приложении
from tasks import process_data

# Асинхронный запуск задачи
result = process_data.delay([1, 2, 3, 4, 5])

# Можно получить результат, когда он будет готов
processed_data = result.get() # Это блокирующий вызов

import dask.dataframe as dd

# Чтение большого CSV-файла, который не помещается в память
df = dd.read_csv('huge_dataset_*.csv')

# Выполнение операций, как с обычным Pandas DataFrame
result = df.groupby('category').agg({'value': 'mean'})

# Вычисления выполняются только при вызове compute()
result.compute()

pip install pyspark

from pyspark.sql import SparkSession

# Инициализация SparkSession — точки входа для всех функций Spark
spark = SparkSession.builder \
.appName("SimpleApp") \
.getOrCreate()

# Чтение данных в DataFrame
df = spark.read.csv("data.csv", header=True, inferSchema=True)

# Выполнение SQL-подобных операций
result = df.filter(df.age > 21) \
.groupBy("department") \
.agg({"salary": "avg"}) \
.orderBy("avg(salary)", ascending=False)

# Показать результаты
result.show()

# Остановка сессии
spark.stop()

from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import col, row_number, avg, sum

spark = SparkSession.builder.appName("AdvancedSpark").getOrCreate()

# Создаем DataFrame с продажами
sales_data = [
("2023-01-01", "Product1", 100, "Store1"),
("2023-01-02", "Product1", 150, "Store2"),
("2023-01-01", "Product2", 200, "Store1"),
("2023-01-03", "Product1", 120, "Store1"),
# ... больше данных
]
df = spark.createDataFrame(sales_data, ["date", "product", "amount", "store"])

# Определяем оконную спецификацию
window_spec = Window.partitionBy("product").orderBy(col("amount").desc())

# Находим топ-продажи для каждого продукта
top_sales = df.withColumn("rank", row_number().over(window_spec)) \
.filter(col("rank") <= 3)

# Агрегация по магазинам и продуктам
summary = df.groupBy("store", "product") \
.agg(sum("amount").alias("total_sales"), avg("amount").alias("avg_sale"))

# Сохранение результатов
top_sales.write.mode("overwrite").parquet("top_sales.parquet")
summary.write.mode("overwrite").csv("sales_summary.csv")

spark.stop()

# Интеграция с ML-библиотеками
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# Подготовка данных для модели
assembler = VectorAssembler(inputCols=["feature1", "feature2", "feature3"], 
outputCol="features")
training_data = assembler.transform(df)

# Разделение на обучающую и тестовую выборки
train_data, test_data = training_data.randomSplit([0\.8, 0.2])

# Обучение модели
rf = RandomForestClassifier(labelCol="label", featuresCol="features")
model = rf.fit(train_data)

# Оценка качества
predictions = model.transform(test_data)
evaluator = MulticlassClassificationEvaluator(
labelCol="label", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print(f"Точность модели: {accuracy}")

pip install dask[complete]

import dask.dataframe as dd
import pandas as pd

# Создание большого тестового датасета
df_sample = pd.DataFrame({
'id': range(1000),
'value': range(1000)
})
df_sample.to_csv('sample.csv', index=False)

# В реальности файл был бы намного больше или их было бы много
df = dd.read_csv('sample*.csv')

# Операции аналогичны Pandas
result = df[df.value > 500].groupby('id').mean()

# Запуск вычислений
result.compute()

import dask.array as da
from dask_ml.linear_model import LogisticRegression
from dask_ml.model_selection import train_test_split

# Создание большого массива данных
X = da.random.random((10000, 50), chunks=(1000, 50))
y = da.random.choice([0, 1], size=10000, chunks=1000)

# Разделение данных на обучающую и тестовую выборки
X_train, X_test, y_train, y_test = train_test_split(X, y)

# Обучение модели логистической регрессии
clf = LogisticRegression()
clf.fit(X_train, y_train)

# Оценка качества
score = clf.score(X_test, y_test)
print(f"Accuracy: {score.compute()}")

# Пример оптимизации работы с Dask
import dask.dataframe as dd
from dask.distributed import Client

# Создание локального кластера
client = Client() # Можно указать n_workers, threads_per_worker и т.д.
print(client.dashboard_link) # URL дашборда для мониторинга

# Чтение данных с оптимизированным размером чанков
df = dd.read_csv('large_data_*.csv', blocksize='64MB')

# Настройка партиционирования для эффективного join
df = df.repartition(npartitions=100)

# Персистенция промежуточных результатов
intermediate = df.query("value > 0").persist()

# Выполнение нескольких операций на персистентных данных
result1 = intermediate.groupby('category').mean()
result2 = intermediate.describe()

# Параллельное вычисление нескольких результатов
from dask import compute
final1, final2 = compute(result1, result2)

import dask
import time

@dask.delayed
def process_chunk(chunk_id):
# Имитация тяжелых вычислений
time.sleep(1)
return chunk_id ** 2

# Создание списка отложенных задач
results = []
for i in range(100):
result = process_chunk(i)
results.append(result)

# Запуск всех задач параллельно
computed_results = dask.compute(*results)
print(f"Результаты: {computed_results}")

# Пример использования паттерна Circuit Breaker
import time
from pybreaker import CircuitBreaker

# Создаем предохранитель, который разомкнется после 5 последовательных ошибок
# и останется разомкнутым на 60 секунд
db_breaker = CircuitBreaker(fail_max=5, reset_timeout=60)

@db_breaker
def query_database(query):
# В реальном коде здесь было бы обращение к БД
if should_fail(): # Эмуляция проблем с БД
raise ConnectionError("Database connection failed")
return execute_query(query)

# Использование с обработкой ошибок
def get_user_data(user_id):
try:
return query_database(f"SELECT * FROM users WHERE id = {user_id}")
except CircuitBreaker.CircuitBreakerError:
# Предохранитель разомкнут, используем резервный источник данных
return get_user_from_cache(user_id)
except ConnectionError:
# Конкретная ошибка подключения, попробуем еще раз позже
return None

# Пример реализации паттерна Bulkhead (переборка) с использованием semaphores
import asyncio
import aiohttp
from functools import wraps

def bulkhead(max_concurrent_calls=10):
semaphore = asyncio.Semaphore(max_concurrent_calls)

def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
async with semaphore:
return await func(*args, **kwargs)
return wrapper
return decorator

# Использование декоратора для ограничения одновременных вызовов
@bulkhead(max_concurrent_calls=5)
async def call_external_service(service_url):
async with aiohttp.ClientSession() as session:
async with session.get(service_url) as response:
return await response.json()

# Реализация паттерна Retry с экспоненциальной задержкой
import time
import random
from functools import wraps

def retry(max_retries=3, base_delay=1, max_delay=60, exceptions=(Exception,)):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while True:
try:
return func(*args, **kwargs)
except exceptions as e:
retries += 1
if retries > max_retries:
raise e

# Экспоненциальная задержка с случайным компонентом
delay = min(base_delay * (2 ** (retries – 1)), max_delay)
jitter = random.uniform(0, 0.1 * delay)
time.sleep(delay + jitter)
return wrapper
return decorator

# Использование
@retry(max_retries=5, exceptions=(ConnectionError, TimeoutError))
def fetch_data_from_api(url):
# Код, который может вызвать исключение
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.json()

# Упрощенная реализация паттерна Saga
class Saga:
def __init__(self):
self.transactions = []
self.compensations = []

def add_transaction(self, transaction, compensation):
self.transactions.append(transaction)
self.compensations.append(compensation)

def execute(self):
current_index = 0
try:
for i, transaction in enumerate(self.transactions):
transaction()
current_index = i + 1
return True
except Exception as e:
# Что-то пошло не так, выполняем компенсирующие действия в обратном порядке
for i in range(current_index – 1, -1, -1):
try:
self.compensations[i]()
except Exception:
# Логирование ошибки в компенсирующей транзакции
pass
# Пробрасываем исходное исключение
raise e

# Пример использования
def book_hotel():
print("Бронируем отель")
return "hotel_booking_id"

def cancel_hotel(booking_id):
print(f"Отменяем бронь отеля {booking_id}")

def book_flight():
print("Бронируем авиабилет")
return "flight_booking_id"

def cancel_flight(booking_id):
print(f"Отменяем авиабилет {booking_id}")

# Создаем сагу для бронирования путешествия
saga = Saga()

# Бронь отеля и компенсация
hotel_id = None
saga.add_transaction(
lambda: globals().update(hotel_id=book_hotel()),
lambda: cancel_hotel(hotel_id)
)

# Бронь авиабилета и компенсация
flight_id = None
saga.add_transaction(
lambda: globals().update(flight_id=book_flight()),
lambda: cancel_flight(flight_id)
)

# Выполнение саги
success = saga.execute()

# Пример интеграции с OpenTelemetry для трассировки распределенных вызовов
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# Настройка трассировки
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# Настройка экспортера Jaeger
jaeger_exporter = JaegerExporter(
agent_host_name="localhost",
agent_port=6831,
)

# Добавление процессора спанов к провайдеру трассировки
trace.get_tracer_provider().add_span_processor(
BatchSpanProcessor(jaeger_exporter)
)

# Использование в коде
@app.route('/api/users/<user_id>')
def get_user(user_id):
with tracer.start_as_current_span("get_user") as span:
span.set_attribute("user.id", user_id)

# Вызов сервиса профиля
with tracer.start_as_current_span("call_profile_service"):
profile = call_profile_service(user_id)

# Вызов сервиса заказов
with tracer.start_as_current_span("call_orders_service"):
orders = call_orders_service(user_id)

return jsonify({"profile": profile, "orders": orders})