Cách tích hợp Apache Spark với Django và MongoDB

Hãy hình dung bạn quản lý một nền tảng thương mại điện tử xử lý hàng nghìn giao dịch mỗi ngày. Bạn muốn phân tích xu hướng bán hàng, theo dõi tăng trưởng doanh thu và dự báo thu nhập tương lai. Các truy vấn cơ sở dữ liệu truyền thống không thể đáp ứng được quy mô hoặc tốc độ này. Vì vậy bạn cần một cách nhanh hơn để xử lý các tập dữ liệu lớn và có được thông tin theo thời gian thực.

Apache Spark lets bạn phân tích khối lượng dữ liệu khổng lồ một cách hiệu quả. Trong hướng dẫn này, tôi sẽ chỉ bạn cách kết nối Django, MongoDB và Apache Spark để phân tích dữ liệu giao dịch thương mại điện tử.

Bạn sẽ thiết lập một dự án Django với MongoDB làm cơ sở dữ liệu và lưu trữ dữ liệu giao dịch trong đó. Sau đó, bạn sẽ dùng PySpark, API Python của Apache Spark, để đọc và lọc dữ liệu. Bạn cũng sẽ thực hiện các phép tính cơ bản và lưu dữ liệu đã xử lý vào MongoDB. Cuối cùng, bạn sẽ hiển thị dữ liệu đã xử lý trong ứng dụng Django của mình.

Để theo dõi tốt nhất hướng dẫn này, bạn nên có hiểu biết cơ bản về Python và framework web Django.

Bây giờ, hãy bắt đầu.

TL;DR

• Sử dụng django-mongodb-backend để kết nối Django với MongoDB Atlas, hỗ trợ gốc cho khóa chính ObjectId
• Cấu hình các ứng dụng tích hợp sẵn của Django (admin, auth, contenttypes) với cấu hình tùy chỉnh để dùng các trường tương thích với MongoDB
• Cài đặt PySpark 3.5.0 với MongoDB Spark Connector 10.4.1 để đọc, lọc và tổng hợp dữ liệu từ MongoDB
• Dùng spark.read.format('mongodb').load() để nạp các collection MongoDB vào PySpark DataFrame
• Ghi dữ liệu đã xử lý trở lại MongoDB bằng .write.format('mongodb').save() và hiển thị kết quả trong template Django

Thiết lập dự án Django

Bắt đầu bằng cách tạo môi trường ảo cho dự án Django của bạn:

python -m venv venv
source venv/bin/activateĐược hỗ trợ bởi

Đảm bảo bạn đã cài đặt Python 3.10 trở lên trong môi trường ảo. Sau đó, cài đặt Django MongoDB Backend:

pip install django-mongodb-backendĐược hỗ trợ bởi

Lệnh trên cũng cài đặt các phiên bản mới nhất của PyMongo 4.x và Django 5.2.x.

Sau khi tải Django MongoDB Backend, hãy tạo dự án Django mới:

django-admin startproject pyspark_tutorialĐược hỗ trợ bởi

Bây giờ, điều hướng vào thư mục dự án và chạy máy chủ phát triển để xác nhận dự án được thiết lập đúng:

cd pyspark_tutorial
python manage.py runserverĐược hỗ trợ bởi

Truy cập http://127.0.0.1:8000/ để xác minh dự án Django đang chạy thành công.

Cấu hình Django để dùng MongoDB

Theo mặc định, Django dùng ID số kiểu AutoField cho khóa chính, phù hợp với cơ sở dữ liệu SQL. Tuy nhiên, MongoDB dùng ObjectId cho ID tài liệu. Để các model tương thích, bạn cần Django tạo khóa chính dạng ObjectId thay vì số nguyên.

Mở pyspark_tutorial/settings.py và cập nhật thiết lập DEFAULT_AUTO_FIELD:

DEFAULT_AUTO_FIELD = 'django_mongodb_backend.fields.ObjectIdAutoField'Được hỗ trợ bởi

Ngay cả với thiết lập toàn cục này, các ứng dụng tích hợp sẵn của Django như admin, auth và contenttypes vẫn mặc định dùng AutoField. Để đảm bảo nhất quán trên tất cả ứng dụng, hãy tạo cấu hình ứng dụng tùy chỉnh để chúng sử dụng ObjectId.

Tạo tệp pyspark_tutorial/apps.py và thêm nội dung sau:

from django.contrib.admin.apps import AdminConfig
from django.contrib.auth.apps import AuthConfig
from django.contrib.contenttypes.apps import ContentTypesConfig


class MongoAdminConfig(AdminConfig):
    default_auto_field = 'django_mongodb_backend.fields.ObjectIdAutoField'

class MongoAuthConfig(AuthConfig):
    default_auto_field = 'django_mongodb_backend.fields.ObjectIdAutoField'

class MongoContentTypesConfig(ContentTypesConfig):
    default_auto_field = 'django_mongodb_backend.fields.ObjectIdAutoField'Được hỗ trợ bởi

Bây giờ, trong pyspark_tutorial/settings.py, cập nhật thiết lập INSTALLED_APPS:

INSTALLED_APPS = [
    'pyspark_tutorial.apps.MongoAdminConfig',
    'pyspark_tutorial.apps.MongoAuthConfig',
    'pyspark_tutorial.apps.MongoContentTypesConfig',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
]Được hỗ trợ bởi

Vì tất cả model phải dùng ObjectIdAutoField, mỗi ứng dụng bên thứ ba và contrib bạn dùng cần có migration riêng tương thích với MongoDB. Vì vậy, hãy thêm nội dung sau vào tệp pyspark_tutorial/setting.py:

MIGRATION_MODULES = {
    'admin': 'mongo_migrations.admin',
    'auth': 'mongo_migrations.auth',
    'contenttypes': 'mongo_migrations.contenttypes',
}Được hỗ trợ bởi

Tạo thư mục mongo_migrations trong thư mục dự án, cùng cấp với tệp manage.py. Cấu trúc thư mục lúc này sẽ giống như sau:

pyspark_tutorial/
├── pyspark_tutorial/
├── mongo_migrations/
└── manage.pyĐược hỗ trợ bởi

Dừng máy chủ bằng Ctrl + C, sau đó tạo migration:

python manage.py makemigrations admin auth contenttypesĐược hỗ trợ bởi

Nếu bạn kiểm tra thư mục mongo_migrations, bạn sẽ thấy một thư mục cho mỗi ứng dụng tích hợp sẵn. Mỗi thư mục chứa các migration của nó.

Để có một mẫu dự án Django đã bao gồm tất cả cấu hình MongoDB như trên, chạy:

django-admin startproject pyspark_tutorial --template https://github.com/mongodb-labs/django-mongodb-project/archive/refs/heads/5.2.x.zipĐược hỗ trợ bởi

Lưu ý: Nếu bạn dùng phiên bản Django khác 5.2.x, hãy thay hai số cho khớp với hai số đầu trong phiên bản của bạn.

Tạo và cấu hình ứng dụng

Bước tiếp theo là tạo một ứng dụng Django để lưu trữ bản ghi giao dịch thô và đã xử lý.

python manage.py startapp salesĐược hỗ trợ bởi

Để cấu hình ứng dụng mới dùng ObjectId, mở sales/apps.py và thay dòng default_auto_field = 'django.db.models.BigAutoField':

from django.apps import AppConfig


class SalesConfig(AppConfig):
    # Use ObjectId as the default primary key field type for MongoDB:
    default_auto_field = 'django_mongodb_backend.fields.ObjectIdAutoField'
    name = 'sales'Được hỗ trợ bởi

Ngoài ra, bạn có thể dùng mẫu startapp sau, vốn đã bao gồm thay đổi ở trên:

python manage.py startapp sales --template https://github.com/mongodb-labs/django-mongodb-app/archive/refs/heads/5.2.x.zipĐược hỗ trợ bởi

Bây giờ, trong pyspark_tutorial/settings.py, thêm ứng dụng sales vào danh sách INSTALLED_APPS:

INSTALLED_APPS = [
    # Add your sales app:
    'sales.apps.SalesConfig',
    ...
]Được hỗ trợ bởi

Tạo triển khai MongoDB

Vì bạn sẽ lưu trữ dữ liệu trong MongoDB, tôi sẽ hướng dẫn bạn cách tạo một triển khai miễn phí trên MongoDB Atlas để lưu trữ và quản lý dữ liệu trên đám mây.

Đăng ký tài khoản Atlas bằng tài khoản Google hoặc địa chỉ email.

Nhấp Create để tạo cụm Free:

Bây giờ, chọn các tùy chọn sau trên trang hiển thị:

• Tùy chọn cụm miễn phí
• Tên cụm: Cluster0
• Nhà cung cấp: AWS
• Vùng: us-east-1

Nhấp Create Deployment:

Bạn sẽ thấy tên người dùng và mật khẩu của mình. Thực hiện như sau:

1. Sao chép tên người dùng và mật khẩu vào tài liệu an toàn.

2. Nhấp Create Database User.

3. Nhấp Choose a connection method để đặt địa chỉ IP kết nối.

Thực hiện như sau trên trang hiển thị:

1. Chọn Drivers.

2. Chọn Python làm driver.

3. Sao chép lệnh trong tùy chọn Install your driver và chạy trong terminal.

4. Sao chép connection string của bạn, bao gồm mật khẩu, trong phần Add your connection string into your application code, và lưu vào tài liệu an toàn.

5. Nhấp Done.

Kết nối Django với MongoDB

Mở pyspark_tutorial/settings.py, và cập nhật thiết lập DATABASES để dùng Django MongoDB Backend với connection string đã lưu. Đồng thời đặt tên cơ sở dữ liệu:

DATABASES = {
    'default': {
        # Change to use Django MongoDB Backend:
        'ENGINE': 'django_mongodb_backend',
        # Use your saved connection string:
        'HOST': '<connection string>',
        # Set a database name:
        'NAME': 'pyspark_tutorial',
    },
}Được hỗ trợ bởi

Trong đoạn mã trên, đảm bảo bạn thay <connection string> bằng connection string đã lưu của bạn.

Cấu hình này kết nối Django với cụm MongoDB Atlas của bạn. ENGINE trỏ đến backend MongoDB, HOST lưu connection string, và NAME định nghĩa tên cơ sở dữ liệu mà Django sẽ sử dụng.

Xây dựng ứng dụng Django

Bây giờ ứng dụng đã được cấu hình, bạn sẽ tạo các model, view, URL và template cần thiết để hiển thị dữ liệu giao dịch trên trình duyệt.

Định nghĩa model

Model mô tả cấu trúc dữ liệu của bạn. Tạo một model để biểu diễn bản ghi giao dịch. Mở sales/models.py và thay thế mã:

from django.db import models

# Define a model to represent each transaction record:
class Transaction(models.Model):
    order_id = models.CharField(max_length=50, unique=True)
    user_id = models.CharField(max_length=50)
    product = models.CharField(max_length=100)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    quantity = models.PositiveIntegerField()
    timestamp = models.DateTimeField()
    country = models.CharField(max_length=50)

    class Meta:
        # Sort transactions by order ID:
        ordering = ['order_id']
        # Add indexes to improve query performance for common lookup fields
        indexes = [
            models.Index(fields=['timestamp']),
            models.Index(fields=['country']),
            models.Index(fields=['product']),
        ]

    def __str__(self):
        return f'{self.order_id} - {self.product}'

    @property
    def total_amount(self):
        # Calculate and return the total transaction amount:
        return self.price * self.quantityĐược hỗ trợ bởi

Model này định nghĩa từng giao dịch với các trường cho thông tin sản phẩm, giá, số lượng và quốc gia. Nó sắp xếp các giao dịch theo order_id. Đồng thời bao gồm index trên các trường chính để cải thiện hiệu năng truy vấn và một thuộc tính tính tổng giá trị giao dịch.

Định nghĩa view

View kiểm soát cách các bản ghi hiển thị trên trình duyệt. Tạo một view để hiển thị các bản ghi giao dịch. Mở sales/views.py và thay thế mã:

from django.shortcuts import render
from .models import Transaction

# Define a view to display all transactions and total revenue:
def transaction_list_view(request):
    # Retrieve all transaction records from the database:
    transactions = Transaction.objects.all()
    # Calculate the total revenue from all transactions:
    total_revenue = sum(t.total_amount for t in transactions)
    # Render the transaction list template with context data
    return render(request, 'sales/transaction_list.html', {
        'transactions': transactions,
        'total_revenue': total_revenue,
    })Được hỗ trợ bởi

View trên lấy tất cả bản ghi giao dịch từ MongoDB bằng model Transaction. Nó tính tổng doanh thu bằng cách cộng trường total_amount của mỗi giao dịch. Sau đó, nó gửi hai giá trị đến một template transaction_list.html mà chúng ta sẽ tạo sau:

* transactions: danh sách tất cả bản ghi giao dịch để hiển thị trong bảng

* total_revenue: tổng số tiền tạo ra từ tất cả giao dịch

Template sẽ dùng các giá trị này để hiển thị từng giao dịch và tổng doanh thu.

Cấu hình URL

Bạn cần tạo tuyến đường để Django biết view nào cần tải khi người dùng truy cập một URL. Tạo tệp sales/urls.py và thêm nội dung sau:

from django.urls import path
from . import views

# Define URL patterns for the sales app:
urlpatterns = [
    # Route the root URL to the transaction list view:
    path('', views.transaction_list_view, name='transaction_list'),
]Được hỗ trợ bởi

Bao gồm URL của ứng dụng vào cấu hình URL của dự án. Mở pyspark_tutorial/urls.py và cập nhật như sau:

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    # Include your app’s URLs:
    path("", include('sales.urls')),
]Được hỗ trợ bởi

Tạo template

Template định nghĩa cách dữ liệu của bạn hiển thị trên trình duyệt. Tạo thư mục sales/templates/sales và thêm tệp tên transaction_list.html bên trong.

Cấu trúc ứng dụng sales của bạn lúc này sẽ giống như sau:

sales/
├── __init__.py
├── admin.py
├── apps.py
├── migrations/
├── models.py
├── views.py
├── tests.py
├── urls.py
└── templates/
   └── sales/
       └── transaction_list.htmlĐược hỗ trợ bởi

Thêm nội dung sau vào tệp transaction_list.html của bạn:

<!DOCTYPE html>
<html>
<head>
    <title>Transactions</title>
</head>
<body>
    <h1>E-commerce Transactions</h1>
    <h3>Total Revenue: ${{ total_revenue }}</h3>

    <table>
        <thead>
            <tr>
                <th>Order ID</th>
                <th>User ID</th>
                <th>Product</th>
                <th>Price</th>
                <th>Quantity</th>
                <th>Country</th>
                <th>Timestamp</th>
            </tr>
        </thead>
        <tbody>
            {% for t in transactions %}
            <tr>
                <td>{{ t.order_id }}</td>
                <td>{{ t.user_id }}</td>
                <td>{{ t.product }}</td>
                <td>${{ t.price }}</td>
                <td>{{ t.quantity }}</td>
                <td>{{ t.country }}</td>
                <td>{{ t.timestamp }}</td>
            </tr>
            {% empty %}
            <tr><td colspan="7">No transactions available.</td></tr>
            {% endfor %}
        </tbody>
    </table>
</body>
</html>Được hỗ trợ bởi

Template trên sử dụng ngôn ngữ template của Django để hiển thị động dữ liệu được truyền từ view.

Ở trên cùng, tổng doanh thu được hiển thị bằng {{ total_revenue }}. Sau đó, template lặp qua tất cả bản ghi giao dịch trong transactions bằng thẻ {% for t in transactions %} và hiển thị mỗi bản ghi như một hàng trong bảng. Mỗi cột hiển thị một thuộc tính cụ thể của giao dịch, như order_id, product, price, quantity và country. Nếu không có bản ghi, thẻ {% empty %} sẽ đảm bảo hiển thị thông báo “No transactions available” thay vì một bảng trống.

Thêm kiểu dáng

Để trang giao dịch của bạn hấp dẫn và dễ đọc hơn, bạn sẽ thêm CSS tùy chỉnh. Django phục vụ các tệp tĩnh như CSS, JavaScript và hình ảnh thông qua một thư mục đặc biệt gọi là static. Điều này giúp tách biệt tệp thiết kế khỏi mã nguồn.

Bây giờ, tạo thư mục sales/static/sales và thêm tệp tên styles.css bên trong. Cấu trúc thư mục sales của bạn sẽ như sau:

sales/
├── __init__.py
├── admin.py
├── apps.py
├── migrations/
├── models.py
├── views.py
├── tests.py
├── urls.py
├── templates/
│   └── sales/
│       └── transaction_list.html
└── static/
   └── sales/
       └── styles.cssĐược hỗ trợ bởi

Trong pyspark_tutorial/settings.py, đảm bảo STATIC_URL được định nghĩa:

STATIC_URL = 'static/'Được hỗ trợ bởi

Thêm nội dung này vào tệp sales/static/sales/styles.css:

table {
  width: 100%;
  border-collapse: collapse;
  margin-top: 20px;
}

th, td {
  border: 1px solid #ddd;
  padding: 8px;
  text-align: left;
}

th {
  background-color: #f5f5f5;
}

body {
  font-family: Arial, sans-serif;
  margin: 20px;
}

h1 {
  color: #333;
}

a {
  text-decoration: none;
  color: #007bff;
}
a:hover {
  text-decoration: underline;
}Được hỗ trợ bởi

Cuối cùng, yêu cầu Django nạp tệp tĩnh và đưa tệp CSS vào template HTML của bạn.

Ở đầu tệp  sales/templates/sales/transaction_list.html, thêm {% load static %},  và sau đó cập nhật phần tử <head> để liên kết đến tệp styles.css:

<!-- Load static files:-->
{% load static %}
<!DOCTYPE html>
<html>
<head>
    <title>Transactions</title>
    <!-- Link the CSS file for styling the template:-->
    <link rel="stylesheet" href="{% static 'sales/styles.css' %}">
</head>Được hỗ trợ bởi

Chạy migration

Tạo và áp dụng migration để Django có thể tạo các collection cần thiết trong MongoDB:

python manage.py makemigrations
python manage.py migrateĐược hỗ trợ bởi

Ghi dữ liệu vào MongoDB

Bây giờ, hãy dùng các model của ứng dụng để thêm bản ghi giao dịch vào cơ sở dữ liệu pyspark_tutorial bằng trình shell tương tác của Django:

python manage.py shellĐược hỗ trợ bởi

Sau đó, import model Transaction từ ứng dụng sales và timezone từ tiện ích của Django:

from sales.models import Transaction
from django.utils import timezoneĐược hỗ trợ bởi

Bây giờ, chèn bản ghi giao dịch bằng bulk_create():

Transaction.objects.bulk_create([
    Transaction(order_id='T1001', user_id='U001', product='Laptop', price=1000.00, quantity=1, timestamp=timezone.now(), country='US'),
    Transaction(order_id='T1002', user_id='U002', product='Smartphone', price=800.00, quantity=2, timestamp=timezone.now(), country='NG'),
    Transaction(order_id='T1003', user_id='U003', product='Headphones', price=150.00, quantity=1, timestamp=timezone.now(), country='UK'),
    Transaction(order_id='T1004', user_id='U004', product='Laptop', price=1200.00, quantity=1, timestamp=timezone.now(), country='US'),
    Transaction(order_id='T1005', user_id='U005', product='Keyboard', price=45.00, quantity=3, timestamp=timezone.now(), country='DE'),
    Transaction(order_id='T1006', user_id='U006', product='Monitor', price=300.00, quantity=2, timestamp=timezone.now(), country='FR'),
    Transaction(order_id='T1007', user_id='U007', product='Smartwatch', price=199.00, quantity=1, timestamp=timezone.now(), country='IN'),
    Transaction(order_id='T1008', user_id='U008', product='Speaker', price=150.00, quantity=2, timestamp=timezone.now(), country='CA'),
    Transaction(order_id='T1009', user_id='U009', product='Camera', price=800.00, quantity=1, timestamp=timezone.now(), country='NG'),
    Transaction(order_id='T1010', user_id='U010', product='Tablet', price=350.00, quantity=1, timestamp=timezone.now(), country='UK'),
    Transaction(order_id='T1011', user_id='U011', product='Headphones', price=75.00, quantity=2, timestamp=timezone.now(), country='US'),
    Transaction(order_id='T1012', user_id='U012', product='Laptop', price=1300.00, quantity=1, timestamp=timezone.now(), country='NG'),
    Transaction(order_id='T1013', user_id='U013', product='Mouse', price=30.00, quantity=3, timestamp=timezone.now(), country='IN'),
    Transaction(order_id='T1014', user_id='U014', product='Smartphone', price=950.00, quantity=1, timestamp=timezone.now(), country='KE'),
    Transaction(order_id='T1015', user_id='U015', product='Keyboard', price=55.00, quantity=2, timestamp=timezone.now(), country='CA'),
    Transaction(order_id='T1016', user_id='U016', product='Smartwatch', price=250.00, quantity=1, timestamp=timezone.now(), country='DE'),
    Transaction(order_id='T1017', user_id='U017', product='Speaker', price=180.00, quantity=1, timestamp=timezone.now(), country='FR'),
    Transaction(order_id='T1018', user_id='U018', product='Monitor', price=400.00, quantity=2, timestamp=timezone.now(), country='NG'),
    Transaction(order_id='T1019', user_id='U019', product='Laptop', price=1250.00, quantity=1, timestamp=timezone.now(), country='IN'),
    Transaction(order_id='T1020', user_id='U020', product='Camera', price=780.00, quantity=1, timestamp=timezone.now(), country='US'),
])Được hỗ trợ bởi

Xác minh rằng các bản ghi đã được thêm thành công:

Transaction.objects.count()Được hỗ trợ bởi

Nếu kết quả trả về 20, các bản ghi của bạn đã được lưu thành công.

Bây giờ, thoát shell bằng cách chạy exit(), và khởi động máy chủ Django để xem giao dịch trên trình duyệt:

python manage.py runserverĐược hỗ trợ bởi

Truy cập http://127.0.0.1:8000/ để xác nhận dữ liệu của bạn được hiển thị.

Xử lý dữ liệu với PySpark

Giờ đây các bản ghi đã được lưu trong cơ sở dữ liệu MongoDB, bạn sẽ dùng Apache Spark để xử lý dữ liệu. Apache Spark đi kèm API Python là PySpark, bạn có thể dùng trong dự án Django để xử lý các tập dữ liệu lớn. Nếu bạn mới làm quen với PySpark, hãy xem hướng dẫn cài đặt PySpark của chúng tôi.

Bạn sẽ tạo một script Python dùng PySpark để kết nối MongoDB và đọc bản ghi giao dịch. Bạn sẽ thực hiện các thao tác cơ bản với dữ liệu này, như lọc. Sau đó, nhóm theo quốc gia và tính tổng doanh thu cho mỗi quốc gia. Cuối cùng, bạn sẽ ghi dữ liệu đã xử lý vào một collection MongoDB mới trong cơ sở dữ liệu.

Cài đặt PySpark

Dừng máy chủ và cài đặt phiên bản PySpark tương thích với MongoDB Spark connector mới nhất:

pip install pyspark==3.5.0Được hỗ trợ bởi

Xác minh cài đặt thành công:

pyspark --versionĐược hỗ trợ bởi

Bạn sẽ nhận được phản hồi chào mừng đến với Spark.

Tạo tệp script PySpark

Tiếp theo, tạo tệp transactions.py trong thư mục dự án, cùng cấp với manage.py. Cấu trúc thư mục của bạn lúc này sẽ như sau:

pyspark_tutorial/
├── mongo_migrations/
├── pyspark_tutorial/
├── sales/
├── manage.py
└── transactions.pyĐược hỗ trợ bởi

Xây dựng connection string

Để kết nối PySpark với MongoDB, bạn phải xây dựng connection string bao gồm tên cơ sở dữ liệu và tên collection của bạn.  Một collection trong MongoDB tương tự như bảng trong cơ sở dữ liệu quan hệ và lưu trữ các tài liệu liên quan. Tên collection của bạn được tạo từ tên ứng dụng và tên model, phân tách bằng dấu gạch dưới (_).

Dùng connection string đã xây dựng trong tệp transactions.py . Định dạng đúng là:

mongodb+srv://<mongodb username>:<mongodb password>@<cluster address>/<database name>.<app name>_<model name>?retryWrites=true&w=majority&appName=Cluster0Được hỗ trợ bởi

Giải thích các giá trị placeholder như sau:

• <mongodb username> và <mongodb password> là tên người dùng và mật khẩu MongoDB Atlas của bạn.
• <cluster address> là miền hiển thị trong connection string Atlas của bạn—ví dụ, cluster0.2rvn82q.mongodb.net.
• <database name> là tên bạn đặt cho cơ sở dữ liệu trong settings.py.
• <app name>_<model name> là tên collection được tạo từ ứng dụng và model Django của bạn—ví dụ, sales_transaction.

Trong phần này, connection string bạn sẽ dùng trong mã PySpark sẽ trông như sau:

mongodb+srv://db_user:[email protected]/pyspark_tutorial.sales_transaction?retryWrites=true&w=majority&appName=Cluster0Được hỗ trợ bởi

Thêm mã PySpark

Thêm đoạn mã sau vào tệp transactions.py:

from pyspark.sql import SparkSession

# Initialize SparkSession with MongoDB connector:
spark = SparkSession.builder \
    .appName('ReadTransactions') \
    .config(
        'spark.mongodb.read.connection.uri',
        # Replace with your built MongoDB connection string:
        '<built connection string>'
    ) \
    .config(
        # Add the MongoDB Spark connector package:
        'spark.jars.packages',
        'org.mongodb.spark:mongo-spark-connector_2.12:10.4.1'
    ) \
    .getOrCreate()

# Read data from MongoDB into a Spark DataFrame:
df = spark.read.format('mongodb').load()

# Show result:
df.show()

# Stop the Spark session:
spark.stop()Được hỗ trợ bởi

from pyspark.sql import SparkSession là điểm vào để sử dụng PySpark. Nó cho phép bạn tương tác với Spark và thực hiện các thao tác dữ liệu.

Dòng spark = SparkSession.builder khởi tạo một phiên Spark mới và đặt tên ReadTransactions. Hai tùy chọn .config() xác định cách Spark kết nối với MongoDB:

• spark.mongodb.read.connection.uri: connection string MongoDB cho Spark biết vị trí cơ sở dữ liệu của bạn
• spark.jars.packages: tải gói MongoDB Spark connector để Spark có thể giao tiếp với MongoDB
• df = spark.read.format('mongodb').load() nạp tất cả tài liệu từ collection được định nghĩa trong connection string vào một PySpark DataFrame, giúp việc truy vấn và biến đổi dữ liệu trở nên dễ dàng.
• .show() in bản xem trước dữ liệu MongoDB ở dạng bảng ngay trong terminal.
• spark.stop() kết thúc ứng dụng Spark và giải phóng tài nguyên hệ thống khi công việc hoàn tất.

Cuối cùng, thay <built connection string> bằng connection string MongoDB đã xây dựng của bạn. Đồng thời, đảm bảo địa chỉ IP hiện tại của bạn đã được đưa vào danh sách cho phép trong danh sách truy cập mạng MongoDB Atlas trước khi kết nối.

Bây giờ, chạy python transactions.py, và bạn sẽ nhận được các bản ghi giao dịch đã lưu từ cơ sở dữ liệu trong terminal.

Lọc dữ liệu với PySpark SQL

Bạn cũng có thể lọc bản ghi bằng các hàm PySpark SQL—ví dụ, chỉ hiển thị giao dịch từ Nigeria (NG). Sửa mã trong tệp transactions.py của bạn:

from pyspark.sql import SparkSession

# Import col to reference DataFrame column names:
from pyspark.sql.functions import col

spark = SparkSession.builder \
    .appName('ReadTransactions') \
    .config(
        'spark.mongodb.read.connection.uri',
        # Replace with your built MongoDB connection string:
        '<built connection string>'
    ) \
    .config(
        'spark.jars.packages',
        'org.mongodb.spark:mongo-spark-connector_2.12:10.4.0'
    ) \
    .getOrCreate()

df = spark.read.format('mongodb').load()

# Filter transactions from Nigeria:
ng_df = df.filter(col('country') == 'NG')

# Show filtered results:
ng_df.show()

# Stop the Spark session:
spark.stop()Được hỗ trợ bởi

Chạy lại python transactions.py, và bạn sẽ thấy các giao dịch từ NG.

Nhóm và tổng hợp dữ liệu

Bạn cũng có thể nhóm bản ghi giao dịch theo quốc gia và tính tổng doanh thu cho mỗi quốc gia. Sửa mã trong transactions.py:

from pyspark.sql import SparkSession
# Import col to reference DataFrame columns and import sum as _sum to avoid naming conflicts:
from pyspark.sql.functions import col, sum as _sum

spark = SparkSession.builder \
    .appName('ReadTransactions') \
    .config(
        'spark.mongodb.read.connection.uri',
        # Replace with your built MongoDB connection string:
        '<Built Connection String>'
    ) \
    .config(
        'spark.jars.packages',
        'org.mongodb.spark:mongo-spark-connector_2.12:10.4.0'
    ) \
    .getOrCreate()

df = spark.read.format('mongodb').load()

# Calculate total revenue per country:
revenue_per_country = df.groupBy('country').agg(
   _sum(col('price') * col('quantity')).alias('total_revenue')
)

# Show total revenue per country:
revenue_per_country.show()

# Stop the Spark session:
spark.stop()Được hỗ trợ bởi

Chạy python transactions.py. Bạn sẽ thấy một bảng hiển thị tổng doanh thu theo quốc gia.

Ghi dữ liệu đã xử lý trở lại MongoDB

Bây giờ bạn đã hiểu cách dùng PySpark để đọc và xử lý dữ liệu, tôi sẽ chỉ bạn cách ghi dữ liệu đã xử lý vào một collection MongoDB.

Ở đây, bạn sẽ ghi DataFrame chứa tổng doanh thu theo quốc gia vào một collection mới.

Sửa mã trong tệp transactions.py:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum

spark = SparkSession.builder \
    .appName('ReadTransactions') \
    .config(
        'spark.mongodb.read.connection.uri',
        # Replace with your built MongoDB connection string:
        '<Built Connection String>'
    ) \
    .config(
        'spark.jars.packages',
        'org.mongodb.spark:mongo-spark-connector_2.12:10.4.0'
    ) \
    .getOrCreate()

df = spark.read.format('mongodb').load()

# Calculate total revenue per country
revenue_per_country = df.groupBy('country').agg(
   _sum(col('price') * col('quantity')).alias('total_revenue')
)

# Write aggregated records into a new MongoDB collection:
revenue_per_country.write \
    .format('mongodb') \
    .mode('overwrite') \
    .option(
        'spark.mongodb.write.connection.uri',
        # Replace with your built MongoDB connection string:
        '<Built Connection String>'
    ) \
    .option(
        # Specify the target database name:
        'spark.mongodb.write.database',
        'pyspark_tutorial'
    ) \
    .option(
        # Specify the target collection name:
        'spark.mongodb.write.collection',
        'revenue_per_country'
    ) \
    .save()

# Stop the Spark session:
spark.stop()Được hỗ trợ bởi

Trong đoạn mã trên, Spark ghi DataFrame đã tổng hợp, revenue_per_country, vào một collection mới tên revenue_per_country trong cơ sở dữ liệu pyspark_tutorial trên MongoDB.

Cuối cùng, hãy đảm bảo bạn thay '<built connection string>' trong cả hai cấu hình kết nối bằng connection string MongoDB đã xây dựng của bạn. Sau đó, chạy python transactions.py. 

Hiển thị dữ liệu đã xử lý trong Django

Bây giờ, hãy hiển thị tổng doanh thu theo quốc gia trên một trang Django.

Đầu tiên, hãy tạo một model mới cho doanh thu theo quốc gia. Thêm đoạn sau vào cuối tệp sales/models.py:

class RevenuePerCountry(models.Model):
    country = models.CharField(max_length=50)
    total_revenue = models.DecimalField(max_digits=15, decimal_places=2)

    class Meta:
        # Define the collection (table) name in MongoDB:
        db_table = 'revenue_per_country'
        # Order results by total_revenue when querying:
        ordering = ['total_revenue']

    def __str__(self):
        # Return a readable string representation of the record:
        return f'{self.country}: ${self.total_revenue}'Được hỗ trợ bởi

Tiếp theo, sửa tệp sales/views.py để lấy và hiển thị collection revenue_per_country:

from django.shortcuts import render
from .models import Transaction

# Import the RevenuePerCountry model:
from .models import RevenuePerCountry

def transaction_list_view(request):
    transactions = Transaction.objects.all()
    total_revenue = sum(t.total_amount for t in transactions)
    return render(request, 'sales/transaction_list.html', {
        'transactions': transactions,
        'total_revenue': total_revenue,
    })

# Add a new view to fetch and display your revenue per country:
def revenue_per_country_view(request):
    revenue_per_country = RevenuePerCountry.objects.all()
    return render(request, 'sales/revenue_per_country.html', {'revenue_per_country': revenue_per_country})Được hỗ trợ bởi

Bao gồm tuyến đường đến view mới trong sales/urls.py:

from django.urls import path
from . import views

urlpatterns = [
    path('', views.transaction_list_view, name='transaction_list'),

    # Add new URL route:
    path('revenue_per_country/', views.revenue_per_country_view, name='revenue_per_country'),
]Được hỗ trợ bởi

Tiếp theo, tạo một template mới để hiển thị tổng doanh thu theo quốc gia.

Trong thư mục sales/templates/sales, tạo tệp revenue_per_country.html và thêm nội dung sau:

{% load static %}
<!DOCTYPE html>
<html lang="en">
<head>
    <title>Transactions</title>
    <link rel="stylesheet" href="{% static 'sales/styles.css' %}">
</head>
<body>
    <h1>Total Revenue by Country</h1>
    <a href="{% url 'transaction_list' %}">← Back to Transactions</a>
    <table>
        <tr>
            <th>Country</th>
            <th>Total Revenue (USD)</th>
        </tr>
        {% for r in revenue_per_country %}
        <tr>
            <td>{{ r.country }}</td>
            <td>${{ r.total_revenue|floatformat:2 }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>Được hỗ trợ bởi

Bạn cũng cần sửa template sales/templates/sales/transaction_list.html để bao gồm liên kết đến trang doanh thu theo quốc gia. Thêm đoạn mã sau sau thẻ <h3> nhưng trước <table>:

<a href="{% url 'revenue_per_country' %}">View Total Revenue by Country</a>Được hỗ trợ bởi

Chạy máy chủ Django:

python manage.py runserverĐược hỗ trợ bởi

Truy cập http://127.0.0.1:8000/revenue_per_country/ để xác nhận trang mới hiển thị đúng.

Chúc mừng! Bạn đã sử dụng thành công Apache Spark để xử lý dữ liệu, lưu vào MongoDB và hiển thị dữ liệu đã xử lý trên một trang web bằng Django.