如何将 Apache Spark 与 Django 和 MongoDB 集成

想象一下，您管理着一个每天处理数千笔交易的电商平台。您希望分析销售趋势、跟踪收入增长并预测未来收益。传统的数据库查询难以在这种规模和速度下胜任。因此，您需要一种更快速的方法来处理大型数据集并获取实时洞察。

Apache Spark 能高效分析海量数据。在本教程中，我将演示如何连接 Django、MongoDB 和 Apache Spark 来分析电商交易数据。

您将使用 MongoDB 作为数据库来搭建一个 Django 项目并在其中存储交易数据。随后，您会使用 PySpark（Apache Spark 的 Python API）来读取和过滤数据。您还将执行基础计算，并将处理后的数据保存到 MongoDB。最后，您会在 Django 应用中展示这些处理结果。

为更好地学习本教程，建议您具备 Python 与 Django Web 框架的基础知识。

现在，我们开始吧。

要点速览

• 使用 django-mongodb-backend 将 Django 连接至 MongoDB Atlas，并原生支持以 ObjectId 作为主键
• 通过自定义 app 配置，让 Django 内置应用（admin、auth、contenttypes）使用与 MongoDB 兼容的字段
• 安装 PySpark 3.5.0 与 MongoDB Spark Connector 10.4.1，从 MongoDB 读取、过滤并聚合数据
• 使用 spark.read.format('mongodb').load() 将 MongoDB 集合加载为 PySpark DataFrame
• 使用 .write.format('mongodb').save() 将处理后的数据写回 MongoDB，并在 Django 模板中展示结果

设置您的 Django 项目

首先为您的 Django 项目创建一个虚拟环境：

python -m venv venv
source venv/bin/activate技术支持

请确保虚拟环境中安装的是 Python 3.10 或更高版本。然后安装 Django MongoDB Backend：

pip install django-mongodb-backend技术支持

上述命令还会安装最新版本的 PyMongo 4.x 与 Django 5.2.x。

下载完 Django MongoDB Backend 后，创建一个新的 Django 项目：

django-admin startproject pyspark_tutorial技术支持

现在进入项目文件夹并运行开发服务器，确认项目配置正确：

cd pyspark_tutorial
python manage.py runserver技术支持

访问 http://127.0.0.1:8000/，验证您的 Django 项目已成功运行。

配置 Django 使用 MongoDB

默认情况下，Django 为主键使用 AutoField 整型 ID，这在 SQL 数据库中很合适。但 MongoDB 的文档 ID 使用 ObjectId。为使模型兼容，您需要让 Django 生成 ObjectId 类型的主键，而不是整数。

打开 pyspark_tutorial/settings.py 并更新 DEFAULT_AUTO_FIELD 设置：

DEFAULT_AUTO_FIELD = 'django_mongodb_backend.fields.ObjectIdAutoField'技术支持

即便使用了上述全局设置，Django 的内置应用如 admin、auth 与 contenttypes 仍会默认使用 AutoField。为确保所有应用一致，创建自定义 app 配置，让它们使用 ObjectId。

创建 pyspark_tutorial/apps.py 文件并添加以下内容：

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'技术支持

现在在 pyspark_tutorial/settings.py 中更新 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',
]技术支持

由于所有模型都必须使用 ObjectIdAutoField，您所用的每个第三方与 contrib 应用都需要针对 MongoDB 的专用迁移。因此在 pyspark_tutorial/setting.py 文件中添加如下内容：

MIGRATION_MODULES = {
    'admin': 'mongo_migrations.admin',
    'auth': 'mongo_migrations.auth',
    'contenttypes': 'mongo_migrations.contenttypes',
}技术支持

在项目文件夹（与 manage.py 同级）中创建一个 mongo_migrations 文件夹。此时的文件结构应类似：

pyspark_tutorial/
├── pyspark_tutorial/
├── mongo_migrations/
└── manage.py技术支持

使用 Ctrl + C 停止服务器，然后生成迁移：

python manage.py makemigrations admin auth contenttypes技术支持

查看 mongo_migrations 文件夹，您会看到每个内置应用都有一个文件夹，并包含对应的迁移。

若想使用已包含上述 MongoDB 配置的 Django 项目模板，请运行：

django-admin startproject pyspark_tutorial --template https://github.com/mongodb-labs/django-mongodb-project/archive/refs/heads/5.2.x.zip技术支持

注意：如果您使用的 Django 版本不是 5.2.x，请将其中两个数字替换为您的版本的前两位数字。

创建并配置您的应用

下一步是创建一个 Django 应用来存储原始与处理后的交易记录。

python manage.py startapp sales技术支持

要将新应用配置为使用 ObjectId，打开 sales/apps.py 并替换 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'技术支持

或者，您也可以使用以下包含上述改动的 startapp 模板：

python manage.py startapp sales --template https://github.com/mongodb-labs/django-mongodb-app/archive/refs/heads/5.2.x.zip技术支持

现在在 pyspark_tutorial/settings.py 中将 sales 应用添加到 INSTALLED_APPS 列表：

INSTALLED_APPS = [
    # Add your sales app:
    'sales.apps.SalesConfig',
    ...
]技术支持

创建一个 MongoDB 部署

由于您将把数据存储在 MongoDB 中，我将向您演示如何在 MongoDB Atlas 上创建免费层部署，以便在云端存储与管理您的数据。

注册 Atlas 账户，可使用 Google 账户或电子邮箱。

点击 Create 来创建一个 Free 集群：

接着，在结果页面选择以下选项：

• 免费集群选项
• 集群名称：Cluster0
• 服务商：AWS
• 区域：us-east-1

点击 Create Deployment：

您会看到您的用户名与密码。请执行以下操作：

1. 将用户名和密码复制到安全文档中。

2. 点击 Create Database User。

3. 点击 Choose a connection method 来设置连接 IP 地址。

在结果页面执行以下操作：

1. 选择 Drivers。

2. 选择 Python 作为驱动。

3. 复制 Install your driver 选项中的命令，在终端中运行。

4. 复制带密码的 connection string（位于 Add your connection string into your application code 下），并保存至安全文档。

5. 点击 Done.

将 Django 连接至 MongoDB

打开 pyspark_tutorial/settings.py，使用您保存的 connection string 配置 Django MongoDB Backend，同时设置数据库名：

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',
    },
}技术支持

在上述代码中，请将 <connection string> 替换为您保存的连接字符串。

该配置将 Django 连接到您的 MongoDB Atlas 集群。 ENGINE 指定 MongoDB 后端，HOST 存储连接字符串，NAME 定义了 Django 将使用的数据库名。

构建 Django 应用

完成应用配置后，您将创建模型、视图、URL 与模板，以在浏览器中展示交易数据。

定义模型

模型描述数据结构。创建一个模型来表示交易记录。打开 sales/models.py 并替换代码：

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技术支持

该模型为每笔交易定义了产品详情、价格、数量与国家等字段，并按 order_id 排序。同时在关键字段上建立索引以提升查询性能，并提供计算交易总额的属性。

定义视图

视图控制这些记录在浏览器中的展示。创建一个视图以显示交易记录。打开 sales/views.py 并替换代码：

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,
    })技术支持

上述视图通过 Transaction 模型从 MongoDB 获取所有交易记录，并累加每笔交易的 total_amount 得到总收入。随后，它向我们稍后将创建的 transaction_list.html 模板传递两个值：

* transactions：将在表格中展示的所有交易记录列表

* total_revenue：所有交易产生的总金额

模板将使用这些值来显示每笔交易与总收入。

配置 URL

您需要创建路由，以便 Django 知道当用户访问某个 URL 时应加载哪个视图。创建 sales/urls.py 文件并添加以下内容：

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'),
]技术支持

在项目 URL 配置中包含应用的 URL。打开 pyspark_tutorial/urls.py 并这样更新：

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')),
]技术支持

创建模板

模板定义数据在浏览器中的呈现方式。创建 sales/templates/sales 文件夹，并在其中添加名为 transaction_list.html 的文件。

您的 sales 应用结构应类似：

sales/
├── __init__.py
├── admin.py
├── apps.py
├── migrations/
├── models.py
├── views.py
├── tests.py
├── urls.py
└── templates/
   └── sales/
       └── transaction_list.html技术支持

在 transaction_list.html 文件中加入以下内容：

<!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>技术支持

上述模板使用了 Django 的模板语言以动态展示来自视图的数据。

页面顶部通过 {{ total_revenue }} 显示总收入。随后模板使用 {% for t in transactions %} 标签遍历所有交易记录，并将每条记录作为表格中的一行展示。每列显示交易的具体属性，如 order_id、product、price、quantity 与 country。若没有记录，{% empty %} 标签会显示 “No transactions available” 的提示，而非空表格。

添加样式

为让交易页面更美观、更易读，您将添加自定义 CSS 样式。Django 通过一个名为 static 的特殊目录来提供静态文件（如 CSS、JavaScript 与图片），这样可将设计文件与代码分离。

现在创建 sales/static/sales 文件夹，并在其中添加 styles.css 文件。此时 sales 文件夹结构如下：

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技术支持

在 pyspark_tutorial/settings.py 中确保已定义 STATIC_URL：

STATIC_URL = 'static/'技术支持

在 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;
}技术支持

最后，告诉 Django 加载静态文件并在 HTML 模板中包含该 CSS 文件。

在 sales/templates/sales/transaction_list.html 顶部添加 {% load static %}，然后在 <head> 元素中链接 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>技术支持

运行迁移

生成并应用迁移，让 Django 能在 MongoDB 中创建所需集合：

python manage.py makemigrations
python manage.py migrate技术支持

向 MongoDB 写入数据

现在，让我们使用应用的模型，通过 Django 交互式 shell 将交易记录添加到 pyspark_tutorial 数据库中：

python manage.py shell技术支持

然后从 sales 应用中导入 Transaction 模型与 Django 工具库中的 timezone：

from sales.models import Transaction
from django.utils import timezone技术支持

现在使用 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'),
])技术支持

验证记录是否成功添加：

Transaction.objects.count()技术支持

如果计数返回 20，说明记录已成功保存。

现在通过运行 exit() 退出 shell，并启动 Django 服务器以在浏览器中查看交易记录：

python manage.py runserver技术支持

访问 http://127.0.0.1:8000/ 确认数据已显示。

使用 PySpark 处理数据

现在交易记录已存储在您的 MongoDB 数据库中，您将使用 Apache Spark 来处理这些数据。Apache Spark 提供了 Python API——PySpark，您可以在 Django 项目中使用它来处理大型数据集。若您是 PySpark 初学者，请查看我们的 PySpark 安装指南。

您将创建一个使用 PySpark 连接 MongoDB 并读取交易记录的 Python 脚本。您会对数据执行基础操作（如过滤），然后按国家分组并计算各自的总收入。最后，您会将处理结果写入数据库中的新 MongoDB 集合。

安装 PySpark

停止服务器并安装与最新MongoDB Spark 连接器兼容的 PySpark 版本：

pip install pyspark==3.5.0技术支持

验证安装是否成功：

pyspark --version技术支持

您应能看到欢迎使用 Spark 的提示。

创建 PySpark 脚本文件

接下来在项目文件夹（与 manage.py 同级）创建 transactions.py 文件。此时文件结构如下：

pyspark_tutorial/
├── mongo_migrations/
├── pyspark_tutorial/
├── sales/
├── manage.py
└── transactions.py技术支持

构建连接字符串

要将 PySpark 连接到 MongoDB，您必须构建一个包含数据库名与集合名的连接字符串。MongoDB 中的集合类似于关系型数据库中的表，用于存储相关文档。集合名由应用名与模型名通过下划线（_）连接而成。

在 transactions.py 文件中使用构建好的连接字符串。正确的格式是：

mongodb+srv://<mongodb username>:<mongodb password>@<cluster address>/<database name>.<app name>_<model name>?retryWrites=true&w=majority&appName=Cluster0技术支持

以下是占位符的说明：

• <mongodb username> 与 <mongodb password>：您的 MongoDB Atlas 用户名与密码。
• <cluster address>：Atlas 连接字符串中显示的域名，例如 cluster0.2rvn82q.mongodb.net。
• <database name>：您在 settings.py 中设置的数据库名。
• <app name>_<model name>：由 Django 应用与模型生成的集合名，例如 sales_transaction。

本节中，您在 PySpark 代码中使用的连接字符串如下：

mongodb+srv://db_user:[email protected]/pyspark_tutorial.sales_transaction?retryWrites=true&w=majority&appName=Cluster0技术支持

添加 PySpark 代码

将以下代码添加到 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()技术支持

from pyspark.sql import SparkSession 是使用 PySpark 的入口，允许您与 Spark 交互并执行数据操作。

通过 spark = SparkSession.builder 初始化新的 Spark 会话并命名为 ReadTransactions。两个 .config() 选项定义了 Spark 如何连接 MongoDB：

• spark.mongodb.read.connection.uri：MongoDB 连接字符串，告诉 Spark 数据库的位置
• spark.jars.packages：下载 MongoDB Spark 连接器包，以便 Spark 与 MongoDB 通信
• df = spark.read.format('mongodb').load() 将连接字符串中定义的集合全部文档加载为 PySpark DataFrame，便于查询与转换
• .show() 在终端中以表格形式预览 MongoDB 数据
• spark.stop() 在任务完成后结束 Spark 应用并释放系统资源

最后，将 <built connection string> 替换为您构建的 MongoDB 连接字符串。在连接前，还请确保您当前的 IP 地址已加入 MongoDB Atlas 网络访问列表 白名单。

现在运行 python transactions.py，您将在终端中看到数据库中的交易记录。

使用 PySpark SQL 过滤数据

您还可以使用 PySpark SQL 函数过滤记录——例如，仅显示来自尼日利亚（NG）的交易。修改 transactions.py 文件中的代码：

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()技术支持

再次运行 python transactions.py，您将看到来自 NG 的交易。

分组与聚合数据

您还可以按国家分组交易记录并计算各自总收入。修改 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()技术支持

运行 python transactions.py。您将看到按国家统计的总收入表。

将处理后的数据写回 MongoDB

了解了如何使用 PySpark 读取并处理数据后，我将向您展示如何将处理结果写入 MongoDB 集合。

这里，您将把包含各国家总收入的 DataFrame 写入一个新集合。

修改 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()技术支持

在上述代码中，Spark 会将聚合得到的 DataFrame（revenue_per_country）写入到 pyspark_tutorial 数据库中的新集合 revenue_per_country。

最后，请确保在两处连接配置中都将 '<built connection string>' 替换为您构建的 MongoDB 连接字符串。然后运行 python transactions.py。

在 Django 中展示处理后的数据

现在，让我们在 Django 页面上展示各国家的总收入。

首先，为各国家收入创建一个新模型。在 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}'技术支持

接着，修改 sales/views.py 文件，以获取并展示 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})技术支持

在 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'),
]技术支持

接下来创建一个新模板，用于展示各国家的总收入。

在 sales/templates/sales 文件夹中创建 revenue_per_country.html 文件并添加以下内容：

{% 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>技术支持

您还需要修改 sales/templates/sales/transaction_list.html 模板，在 <h3> 之后、<table> 之前加入指向各国家收入页面的链接：

<a href="{% url 'revenue_per_country' %}">View Total Revenue by Country</a>技术支持

运行 Django 服务器：

python manage.py runserver技术支持

访问 http://127.0.0.1:8000/revenue_per_country/ 确认新页面显示正常。

恭喜！您已成功使用 Apache Spark 处理数据，将结果存储到 MongoDB，并通过 Django 在网页上展示处理后的数据。