Cara Mengintegrasikan Apache Spark Dengan Django dan MongoDB

Bayangkan Anda mengelola platform e-commerce yang memproses ribuan transaksi setiap hari. Anda ingin menganalisis tren penjualan, melacak pertumbuhan pendapatan, dan memproyeksikan pemasukan di masa mendatang. Kueri basis data tradisional tidak mampu menangani skala atau kecepatan ini. Jadi Anda memerlukan cara yang lebih cepat untuk memproses kumpulan data besar dan mendapatkan wawasan secara real time.

Apache Spark lebihkan Anda menganalisis data berukuran masif secara efisien. Dalam tutorial ini, saya akan menunjukkan cara menghubungkan Django, MongoDB, dan Apache Spark untuk menganalisis data transaksi e-commerce.

Anda akan menyiapkan proyek Django dengan MongoDB sebagai basis data dan menyimpan data transaksi di dalamnya. Lalu, Anda akan menggunakan PySpark, API Python untuk Apache Spark, untuk membaca dan memfilter data. Anda juga akan melakukan perhitungan dasar dan menyimpan data yang telah diproses di MongoDB. Terakhir, Anda akan menampilkan data yang telah diproses di aplikasi Django Anda.

Agar mendapatkan hasil terbaik dari tutorial ini, Anda perlu memahami dasar-dasar Python dan kerangka kerja web Django.

Sekarang, mari kita mulai.

Ringkasan

• Gunakan django-mongodb-backend untuk menghubungkan Django ke MongoDB Atlas dengan dukungan native untuk kunci utama ObjectId
• Konfigurasikan aplikasi bawaan Django (admin, auth, contenttypes) dengan konfigurasi aplikasi kustom agar menggunakan field yang kompatibel dengan MongoDB
• Instal PySpark 3.5.0 dengan MongoDB Spark Connector 10.4.1 untuk membaca, memfilter, dan mengagregasi data dari MongoDB
• Gunakan spark.read.format('mongodb').load() untuk memuat koleksi MongoDB ke dalam DataFrame PySpark
• Tulis data yang telah diproses kembali ke MongoDB menggunakan .write.format('mongodb').save() dan tampilkan hasilnya di template Django

Siapkan Proyek Django Anda

Mulailah dengan membuat virtual environment untuk proyek Django Anda:

python -m venv venv
source venv/bin/activate

Pastikan Anda telah memasang Python 3.10 atau yang lebih baru di virtual environment Anda. Lalu, instal Django MongoDB Backend:

pip install django-mongodb-backend

Perintah di atas juga memasang versi terbaru PyMongo 4.x dan Django 5.2.x.

Setelah Anda mengunduh Django MongoDB Backend, buat proyek Django baru:

django-admin startproject pyspark_tutorial

Sekarang, navigasikan ke folder proyek dan jalankan server pengembangan untuk memastikan proyek Anda tersetup dengan benar:

cd pyspark_tutorial
python manage.py runserver

Kunjungi http://127.0.0.1:8000/ untuk memverifikasi bahwa proyek Django Anda berhasil berjalan.

Konfigurasikan Django untuk Menggunakan MongoDB

Secara default, Django menggunakan ID integer AutoField untuk primary key, yang cocok untuk basis data SQL. Namun, MongoDB menggunakan ObjectId untuk ID dokumen. Agar model Anda kompatibel, Anda perlu membuat Django menghasilkan primary key sebagai ObjectId alih-alih integer.

Buka pyspark_tutorial/settings.py dan perbarui pengaturan DEFAULT_AUTO_FIELD:

DEFAULT_AUTO_FIELD = 'django_mongodb_backend.fields.ObjectIdAutoField'

Meski dengan pengaturan global ini, aplikasi bawaan Django seperti admin, auth, dan contenttypes masih akan menggunakan default AutoField. Untuk memastikan konsistensi di semua aplikasi, buat konfigurasi aplikasi kustom agar menggunakan ObjectId.

Buat file pyspark_tutorial/apps.py dan tambahkan berikut ini:

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'

Sekarang, di pyspark_tutorial/settings.py, perbarui pengaturan INSTALLED_APPS Anda:

INSTALLED_APPS = [
    'pyspark_tutorial.apps.MongoAdminConfig',
    'pyspark_tutorial.apps.MongoAuthConfig',
    'pyspark_tutorial.apps.MongoContentTypesConfig',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
]

Karena semua model harus menggunakan ObjectIdAutoField, setiap aplikasi pihak ketiga dan contrib yang Anda gunakan perlu memiliki migration khusus untuk MongoDB. Jadi tambahkan berikut ini ke file pyspark_tutorial/setting.py Anda:

MIGRATION_MODULES = {
    'admin': 'mongo_migrations.admin',
    'auth': 'mongo_migrations.auth',
    'contenttypes': 'mongo_migrations.contenttypes',
}

Buat folder mongo_migrations di folder proyek Anda, sejajar dengan file manage.py. Struktur folder Anda pada tahap ini seharusnya seperti ini:

pyspark_tutorial/
├── pyspark_tutorial/
├── mongo_migrations/
└── manage.py

Hentikan server dengan Ctrl + C, lalu buat migration Anda:

python manage.py makemigrations admin auth contenttypes

Jika Anda memeriksa folder mongo_migrations, Anda akan melihat folder untuk setiap aplikasi bawaan. Masing-masing folder berisi migration-nya.

Untuk template proyek Django yang sudah memiliki semua konfigurasi MongoDB di atas, jalankan:

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

Catatan: Jika Anda menggunakan versi Django selain 5.2.x, ganti dua angka tersebut agar sesuai dengan dua angka pertama dari versi Anda.

Buat dan Konfigurasikan Aplikasi Anda

Langkah berikutnya adalah membuat aplikasi Django untuk menyimpan catatan transaksi mentah dan yang telah diproses.

python manage.py startapp sales

Untuk mengonfigurasi aplikasi baru Anda agar menggunakan ObjectId, buka sales/apps.py dan ganti baris 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'

Sebagai alternatif, Anda dapat menggunakan template startapp berikut yang sudah mencakup perubahan di atas:

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

Sekarang, di pyspark_tutorial/settings.py, tambahkan aplikasi sales Anda ke daftar INSTALLED_APPS:

INSTALLED_APPS = [
    # Add your sales app:
    'sales.apps.SalesConfig',
    ...
]

Buat Deployment MongoDB

Karena Anda akan menyimpan data di MongoDB, saya akan memandu Anda cara membuat deployment gratis di MongoDB Atlas untuk menyimpan dan mengelola data Anda di cloud.

Daftarkan akun Atlas menggunakan akun Google atau alamat email Anda.

Klik Create untuk membuat cluster Free:

Sekarang, pilih opsi berikut di halaman yang muncul:

• Opsi cluster gratis
• Nama cluster: Cluster0
• Penyedia: AWS
• Wilayah: us-east-1

Klik Create Deployment:

Anda akan melihat nama pengguna dan kata sandi Anda. Lakukan hal berikut:

1. Salin nama pengguna dan kata sandi Anda ke dokumen yang aman.

2. Klik Create Database User.

3. Klik Choose a connection method untuk menetapkan alamat IP koneksi.

Lakukan hal berikut di halaman yang muncul:

1. Pilih Drivers.

2. Pilih Python sebagai driver Anda.

3. Salin perintah pada opsi Install your driver dan jalankan di terminal Anda.

4. Salin connection string Anda, termasuk kata sandi, di bawah Add your connection string into your application code, dan simpan di dokumen yang aman.

5. Klik Done.

Hubungkan Django ke MongoDB

Buka pyspark_tutorial/settings.py, dan perbarui pengaturan DATABASES untuk menggunakan Django MongoDB Backend dengan connection string yang telah Anda simpan. Tetapkan juga nama basis data:

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',
    },
}

Pada kode di atas, pastikan Anda mengganti <connection string> dengan connection string yang telah Anda simpan.

Konfigurasi ini menghubungkan Django ke cluster MongoDB Atlas Anda. ENGINE mengarah ke backend MongoDB, HOST menyimpan connection string Anda, dan NAME mendefinisikan nama basis data yang akan digunakan Django.

Bangun Aplikasi Django

Sekarang setelah aplikasi Anda terkonfigurasi, Anda akan membuat model, view, URL, dan template yang diperlukan untuk menampilkan data transaksi di browser.

Definisikan model

Model menjelaskan struktur data Anda. Buat model untuk merepresentasikan catatan transaksi. Buka sales/models.py dan ganti kodenya:

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

Model ini mendefinisikan setiap transaksi dengan field untuk detail produk, harga, kuantitas, dan negara. Transaksi diurutkan berdasarkan order_id. Model ini juga menyertakan indeks pada field kunci untuk meningkatkan performa kueri dan properti yang menghitung total nilai transaksi.

Definisikan view

View mengontrol bagaimana catatan tersebut tampil di browser. Buat view untuk menampilkan catatan transaksi. Buka sales/views.py dan ganti kodenya:

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,
    })

View di atas mengambil semua catatan transaksi dari MongoDB menggunakan model Transaction. View ini menghitung total pendapatan dengan menjumlahkan field total_amount dari setiap transaksi. Kemudian, view mengirim dua nilai ke template transaction_list.html yang akan kita buat nanti:

* transactions: daftar semua catatan transaksi untuk ditampilkan dalam tabel

* total_revenue: total nilai yang dihasilkan dari semua transaksi

Template akan menggunakan nilai-nilai ini untuk menampilkan setiap transaksi dan total pendapatan.

Konfigurasikan URL

Anda perlu membuat rute agar Django mengetahui view mana yang harus dimuat saat pengguna mengunjungi sebuah URL. Buat file sales/urls.py dan tambahkan berikut ini:

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'),
]

Sertakan URL aplikasi Anda dalam konfigurasi URL proyek. Buka pyspark_tutorial/urls.py dan perbarui seperti ini:

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')),
]

Buat template

Template menentukan bagaimana data Anda tampil di browser. Buat folder sales/templates/sales dan tambahkan file bernama transaction_list.html di dalamnya.

Struktur aplikasi sales Anda seharusnya seperti ini:

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

Tambahkan berikut ini ke file transaction_list.html Anda:

<!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>

Template di atas menggunakan bahasa templating Django untuk menampilkan data secara dinamis yang dikirim dari view.

Di bagian atas, total pendapatan ditampilkan menggunakan {{ total_revenue }}. Lalu, template melakukan iterasi pada semua catatan transaksi dalam transactions menggunakan tag {% for t in transactions %} dan menampilkan setiap catatan sebagai baris pada tabel. Setiap kolom menampilkan atribut tertentu dari sebuah transaksi, seperti order_id, product, price, quantity, dan country. Jika tidak ada catatan, tag {% empty %} memastikan bahwa pesan “No transactions available” muncul alih-alih tabel kosong.

Tambahkan style

Agar halaman transaksi Anda lebih menarik secara visual dan mudah dibaca, Anda akan menambahkan CSS kustom. Django menyajikan file statis, seperti CSS, JavaScript, dan gambar, melalui direktori khusus bernama static. Ini menjaga file desain terpisah dari kode Anda.

Sekarang, buat folder sales/static/sales dan tambahkan file bernama styles.css di dalamnya. Struktur folder sales Anda seharusnya terlihat seperti ini:

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

Di pyspark_tutorial/settings.py, pastikan STATIC_URL didefinisikan:

STATIC_URL = 'static/'

Tambahkan ini ke file sales/static/sales/styles.css Anda:

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;
}

Terakhir, beri tahu Django untuk memuat file statis dan sertakan file CSS di template HTML Anda.

Di bagian atas sales/templates/sales/transaction_list.html, tambahkan {% load static %}, lalu perbarui elemen <head> untuk menautkan ke file 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>

Jalankan migration

Buat dan terapkan migration agar Django dapat membuat koleksi yang diperlukan di MongoDB:

python manage.py makemigrations
python manage.py migrate

Tulis Data ke MongoDB

Sekarang, mari gunakan model aplikasi Anda untuk menambahkan catatan transaksi ke basis data pyspark_tutorial menggunakan shell interaktif Django:

python manage.py shell

Lalu, impor model Transaction dari aplikasi sales Anda dan timezone dari utilitas Django:

from sales.models import Transaction
from django.utils import timezone

Sekarang, sisipkan catatan transaksi menggunakan 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'),
])

Verifikasi bahwa catatan berhasil ditambahkan:

Transaction.objects.count()

Jika jumlahnya mengembalikan 20, catatan Anda berhasil disimpan.

Sekarang, keluar dari shell dengan menjalankan exit(), dan jalankan server Django untuk melihat transaksi Anda di browser:

python manage.py runserver

Kunjungi http://127.0.0.1:8000/ untuk memastikan data Anda ditampilkan.

Proses Data dengan PySpark

Sekarang setelah catatan Anda tersimpan di basis data MongoDB, Anda akan menggunakan Apache Spark untuk memproses data. Apache Spark dilengkapi API Python, PySpark, yang dapat Anda gunakan di proyek Django Anda untuk memproses kumpulan data besar. Jika Anda baru mengenal PySpark, lihat panduan instalasi PySpark kami.

Anda akan membuat skrip Python yang menggunakan PySpark untuk terhubung ke MongoDB dan membaca catatan transaksi Anda. Anda akan melakukan operasi dasar pada data ini, seperti memfilternya. Lalu, Anda akan mengelompokkan berdasarkan negara dan menghitung total pendapatan untuk masing-masing. Terakhir, Anda akan menulis data yang telah diproses ke koleksi MongoDB baru di basis data Anda.

Instal PySpark

Hentikan server Anda dan instal versi PySpark yang kompatibel dengan MongoDB Spark connector terbaru:

pip install pyspark==3.5.0

Verifikasi bahwa instalasi berhasil:

pyspark --version

Anda akan mendapatkan respons sambutan ke Spark.

Buat file skrip PySpark

Selanjutnya, buat file transactions.py di folder proyek Anda, sejajar dengan manage.py. Struktur folder Anda sekarang seharusnya seperti ini:

pyspark_tutorial/
├── mongo_migrations/
├── pyspark_tutorial/
├── sales/
├── manage.py
└── transactions.py

Bangun connection string Anda

Untuk menghubungkan PySpark ke MongoDB, Anda harus membangun connection string yang mencakup nama basis data dan nama koleksi Anda. Koleksi di MongoDB mirip dengan tabel pada basis data relasional dan menyimpan dokumen terkait. Nama koleksi Anda tersusun dari nama aplikasi dan nama model, dipisahkan oleh garis bawah (_).

Gunakan connection string yang dibuat di file transactions.py Anda. Format yang benar adalah:

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

Berikut penjelasan nilai placeholder:

• <mongodb username> dan <mongodb password> adalah nama pengguna dan kata sandi MongoDB Atlas Anda.
• <cluster address> adalah domain yang ditampilkan dalam connection string Atlas Anda—misalnya, cluster0.2rvn82q.mongodb.net.
• <database name> adalah nama yang Anda berikan ke basis data Anda di settings.py.
• <app name>_<model name> adalah nama koleksi yang dihasilkan dari aplikasi dan model Django Anda—misalnya, sales_transaction.

Pada bagian ini, string koneksi yang akan Anda gunakan dalam kode PySpark Anda akan terlihat seperti ini:

mongodb+srv://db_user:password@cluster.mongodb.net/pyspark_tutorial.sales_transaction?retryWrites=true&w=majority&appName=Cluster0

Tambahkan kode PySpark

Tambahkan kode berikut ke file transactions.py Anda:

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 adalah titik masuk untuk menggunakan PySpark. Ini memungkinkan Anda berinteraksi dengan Spark dan melakukan operasi data.

Baris spark = SparkSession.builder menginisialisasi sesi Spark baru dan memberinya nama, ReadTransactions. Dua opsi .config() menentukan cara Spark harus terhubung ke MongoDB:

• spark.mongodb.read.connection.uri: string koneksi MongoDB yang memberi tahu Spark lokasi basis data Anda
• spark.jars.packages: mengunduh paket MongoDB Spark connector agar Spark dapat berkomunikasi dengan MongoDB
• df = spark.read.format('mongodb').load() memuat semua dokumen dari koleksi yang didefinisikan dalam string koneksi Anda ke dalam PySpark DataFrame, sehingga memudahkan untuk melakukan kueri dan transformasi data.
• .show() mencetak pratinjau data MongoDB Anda dalam format tabel langsung di terminal.
• spark.stop() mengakhiri aplikasi Spark dan melepaskan sumber daya sistem setelah pekerjaan selesai.

Terakhir, ganti <built connection string> dengan string koneksi MongoDB yang sudah Anda buat. Pastikan juga alamat IP Anda saat ini masuk daftar putih di daftar akses jaringan MongoDB Atlas sebelum melakukan koneksi.

Sekarang, jalankan python transactions.py, dan Anda akan mendapatkan catatan transaksi yang tersimpan dari basis data Anda di terminal.

Filter data dengan PySpark SQL

Anda juga dapat memfilter catatan Anda menggunakan fungsi PySpark SQL—misalnya, untuk menampilkan hanya transaksi dari Nigeria (NG). Ubah kode dalam file transactions.py Anda:

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()

Jalankan python transactions.py lagi, dan Anda akan melihat transaksi dari NG.

Kelompokkan dan agregasikan data

Anda juga dapat mengelompokkan catatan transaksi berdasarkan negara dan menghitung total pendapatan untuk masing-masing negara. Ubah kode dalam transactions.py Anda:

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()

Jalankan python transactions.py. Anda akan melihat tabel yang menampilkan total pendapatan per negara.

Tulis data yang telah diproses kembali ke MongoDB

Sekarang setelah Anda memahami cara menggunakan PySpark untuk membaca dan memproses data, saya akan menunjukkan cara menulis data yang telah diproses ke dalam koleksi MongoDB.

Di sini, Anda akan menulis DataFrame yang berisi total pendapatan per negara ke dalam koleksi baru.

Ubah kode dalam file transactions.py Anda:

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()

Pada kode di atas, Spark menulis DataFrame teragregasi, revenue_per_country, ke koleksi baru bernama revenue_per_country di basis data pyspark_tutorial Anda di MongoDB.

Terakhir, pastikan Anda mengganti '<built connection string>' pada kedua konfigurasi koneksi dengan string koneksi MongoDB yang sudah Anda buat. Lalu, jalankan python transactions.py. 

Tampilkan Data yang Diproses di Django

Sekarang, mari tampilkan total pendapatan per negara pada halaman Django.

Pertama, mari buat model baru untuk pendapatan per negara. Tambahkan ini di bagian bawah file sales/models.py Anda:

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}'

Selanjutnya, ubah file sales/views.py Anda untuk mengambil dan menampilkan koleksi 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})

Tambahkan rute ke view baru di 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'),
]

Selanjutnya, buat template baru untuk menampilkan total pendapatan per negara.

Di folder sales/templates/sales, buat file revenue_per_country.html dan tambahkan berikut ini:

{% 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>

Anda juga perlu mengubah template sales/templates/sales/transaction_list.html untuk menyertakan tautan ke halaman pendapatan per negara. Tambahkan kode berikut setelah <h3> tetapi sebelum <table>:

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

Jalankan server Django:

python manage.py runserver

Kunjungi http://127.0.0.1:8000/revenue_per_country/ untuk memastikan halaman baru ditampilkan dengan benar.

Selamat! Anda berhasil menggunakan Apache Spark untuk memproses data, menyimpannya di MongoDB, dan menampilkan data yang telah diproses pada halaman web menggunakan Django.