Daily Python Knowledge Point: Mastering Distributed System Construction and Mainstream Framework Practice

Today’s Knowledge Point: Mastering Distributed System Construction and Mainstream Framework Practice

When single-machine performance reaches its limits, distributed computing is the key technology to break through these constraints. Today, we will build a complete distributed system, from task queues to big data processing, unlocking Python’s cluster computing capabilities!

Core Concepts of Distributed Computing

Concept	Description	Typical Implementation
Task Distribution	Main node assigns tasks	Celery, Dask
Data Parallelism	Sharded processing of datasets	PySpark
Message Passing	Communication between nodes	Redis/RabbitMQ
Fault Tolerance Mechanism	Retrying failed tasks	Apache Airflow

Celery: Distributed Task Queue

Infrastructure

# tasks.py
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def process_data(data):    # Time-consuming task processing    return result.upper()

Starting Worker

celery -A tasks worker --loglevel=info

Task Scheduling

from tasks import process_data
# Asynchronous execution
result = process_data.delay("hello world")
# Get result
print(result.get(timeout=10))  # "HELLO WORLD"

Scheduled Tasks

app.conf.beat_schedule = {    'every-10-seconds': {        'task': 'tasks.process_data',        'schedule': 10.0,        'args': ('scheduled task',)    },}

Dask: Flexible Parallel Computing

Data Set Parallelization

import dask.array as da
# Create a 10 billion element array (virtual allocation)
x = da.random.random((100000, 100000), chunks=(1000, 1000))
# Distributed computation
result = (x + x.T).mean()
print(result.compute())  # Trigger actual computation

Distributed Cluster

from dask.distributed import Client
# Connect to cluster
client = Client("tcp://scheduler:8786")
# Distributed execution
futures = []
for url in urls:    future = client.submit(process_url, url)    futures.append(future)
results = client.gather(futures)

PySpark: Big Data Processing

Basic RDD Operations

from pyspark import SparkContext
sc = SparkContext("local", "WordCount")
# Text processing
text_rdd = sc.textFile("huge_file.txt")
word_counts = text_rdd.flatMap(lambda line: line.split()) \
                     .map(lambda word: (word, 1)) \
                     .reduceByKey(lambda a, b: a + b)
# Save results
word_counts.saveAsTextFile("output")

DataFrame API

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("DataAnalysis").getOrCreate()
df = spark.read.csv("data.csv", header=True)
result = df.filter(df.age > 30) \
           .groupBy("department") \
           .agg({"salary": "avg", "age": "max"})
result.show()

Distributed System Communication

Redis Publish/Subscribe

import redis
# Publisher
r = redis.Redis()
r.publish('channel', 'message')
# Subscriber
pubsub = r.pubsub()
pubsub.subscribe('channel')
for message in pubsub.listen():    if message['type'] == 'message':        print(message['data'])

ZeroMQ Advanced Patterns

import zmq
# Request-Response pattern
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")
while True:    message = socket.recv_string()    socket.send_string(f"Processing: {message}")

Fault Tolerance and Monitoring

Task Retry Mechanism

@app.task(bind=True, max_retries=3)
def unreliable_task(self):    try:        # Operation that may fail        return api_call()    except Exception as exc:        self.retry(exc=exc, countdown=2**self.request.retries)

Distributed Monitoring

# Flower monitoring Celery
celery -A tasks flower --port=5555
# Access http://localhost:5555

Practical: Distributed Image Processing System

# Architecture Design
"""
1. Client uploads image to S3
2. Main node generates processing tasks
3. Worker nodes process in parallel
4. Results stored in database
"""
# Task Definition
@app.task
def process_image(image_key):    from PIL import Image    import boto3
    s3 = boto3.client('s3')
    # Download image
    s3.download_file('my-bucket', image_key, 'local.jpg')
    # Process image
    img = Image.open('local.jpg')    img = img.rotate(45).filter(ImageFilter.BLUR)    img.save('processed.jpg')
    # Upload result
    s3.upload_file('processed.jpg', 'my-bucket', f'processed/{image_key}')
    return f"s3://my-bucket/processed/{image_key}"
# Client call
result = process_image.delay("uploads/photo1.jpg")
print(result.get())

Distributed Deployment Solutions

Docker Containerization

# Celery Worker Dockerfile
FROM python:3.9
RUN pip install celery redis pillow
COPY tasks.py .
CMD ["celery", "-A", "tasks", "worker", "--loglevel=info"]

Kubernetes Orchestration

# Kubernetes Deployment Configuration
apiVersion: apps/v1
kind: Deployment
metadata:  name: celery-worker
spec:  replicas: 5  # 5 Worker instances  selector:    matchLabels:      app: celery  template:    metadata:      labels:        app: celery    spec:      containers:      - name: worker        image: my-celery-image:v1        env:        - name: BROKER_URL          value: "redis://redis-service:6379/0"

Today’s Summary

Distributed Architecture: Three main patterns of task distribution, data parallelism, and message passing
Celery Practice: Building a reliable distributed task queue system
Dask Application: Techniques for parallel processing of large datasets
PySpark Ecosystem: Big data processing on Hadoop clusters
Communication Mechanisms: Redis/ZeroMQ for inter-node communication
Fault Tolerance Monitoring: Task retry and cluster monitoring solutions
Deployment Solutions: Docker + Kubernetes containerized deployment

Leave a Comment Cancel reply