Apache Spark'ı Django ve MongoDB ile Nasıl Entegre Edebilirsiniz

Günde binlerce işlemi işleyen bir e-ticaret platformunu yönettiğinizi hayal edin. Satış trendlerini analiz etmek, gelir artışını izlemek ve gelecekteki geliri tahmin etmek istiyorsunuz. Geleneksel veritabanı sorguları bu ölçeği veya hızı kaldıramaz. Bu nedenle, büyük veri kümelerini daha hızlı işleyip gerçek zamanlı içgörüler elde etmenin bir yoluna ihtiyacınız var.

Apache Spark large veri hacimlerini verimli bir şekilde analiz etmenizi sağlar. Bu eğitimde, Django, MongoDB ve Apache Spark'ı bağlayarak e-ticaret işlem verilerini nasıl analiz edeceğinizi

MongoDB'yi veritabanı olarak kullanan bir Django projesi kuracak ve işlem verilerini burada depolayacaksınız. Ardından, PySpark'ı, yani Apache Spark için Python API'sini kullanarak verileri okuyup filtreleyeceksiniz. Ayrıca temel hesaplamalar yapacak ve işlenmiş verileri MongoDB'ye kaydedeceksiniz. Son olarak, işlenmiş verileri Django uygulamanızda görüntüleyeceksiniz.

Bu eğitimden en iyi şekilde yararlanmak için Python ve Django web çatısı hakkında temel düzeyde bilgi sahibi olmalısınız.

Şimdi başlayalım.

Özet

• django-mongodb-backend ile Django'yu MongoDB Atlas'a, ObjectId birincil anahtarlar için yerel destekle bağlayın
• Django'nun yerleşik uygulamalarını (admin, auth, contenttypes) MongoDB uyumlu alanlar kullanacak şekilde özel uygulama yapılandırmalarıyla düzenleyin
• MongoDB'den veri okumak, filtrelemek ve toplamak için MongoDB Spark Connector 10.4.1 ile PySpark 3.5.0'ı kurun
• MongoDB koleksiyonlarını PySpark DataFrame'lerine yüklemek için spark.read.format('mongodb').load() kullanın
• İşlenmiş verileri .write.format('mongodb').save() ile MongoDB'ye geri yazın ve sonuçları Django şablonlarında görüntüleyin

Django Projenizi Kurun

Sanal bir ortam oluşturarak başlayın:

python -m venv venv
source venv/bin/activate

Sanal ortamınızda Python 3.10 veya üzeri kurulu olduğundan emin olun. Ardından, Django MongoDB Backend'i kurun:

pip install django-mongodb-backend

Yukarıdaki komut ayrıca en yeni PyMongo 4.x ve Django 5.2.x sürümlerini de yükler.

Django MongoDB Backend'i indirdikten sonra yeni bir Django projesi oluşturun:

django-admin startproject pyspark_tutorial

Şimdi proje klasörüne gidin ve geliştirme sunucusunu çalıştırarak kurulumun doğru olduğunu doğrulayın:

cd pyspark_tutorial
python manage.py runserver

Django projenizin başarıyla çalıştığını doğrulamak için http://127.0.0.1:8000/ adresini ziyaret edin.

Django'yu MongoDB Kullanacak Şekilde Yapılandırın

Varsayılan olarak Django, birincil anahtarlar için AutoField tamsayı kimlikleri kullanır; bu SQL veritabanlarıyla iyi çalışır. Ancak MongoDB belge kimlikleri için ObjectId kullanır. Modellerinizi uyumlu hale getirmek için Django'nun birincil anahtarları tamsayı yerine ObjectId olarak oluşturması gerekir.

pyspark_tutorial/settings.py dosyasını açın ve DEFAULT_AUTO_FIELD ayarını güncelleyin:

DEFAULT_AUTO_FIELD = 'django_mongodb_backend.fields.ObjectIdAutoField'

Bu genel ayara rağmen, admin, auth ve contenttypes gibi Django'nun yerleşik uygulamaları varsayılan olarak yine AutoField kullanır. Tüm uygulamalarda tutarlılığı sağlamak için bu uygulamaların ObjectId kullanmasını sağlayacak özel uygulama yapılandırmaları oluşturun.

pyspark_tutorial/apps.py dosyası oluşturun ve şunları ekleyin:

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'

Şimdi, pyspark_tutorial/settings.py içinde INSTALLED_APPS ayarını güncelleyin:

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

Tüm modellerin ObjectIdAutoField kullanması gerektiğinden, kullandığınız her üçüncü taraf ve contrib uygulamasının MongoDB'ye özel kendi geçişlerine sahip olması gerekir. Bu nedenle pyspark_tutorial/setting.py dosyanıza şunları ekleyin:

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

Proje klasörünüzde, manage.py dosyanızla aynı seviyede bir mongo_migrations klasörü oluşturun. Bu aşamadaki klasör yapınız şu şekilde olmalı:

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

Ctrl + C ile sunucuyu durdurun, ardından geçişleri oluşturun:

python manage.py makemigrations admin auth contenttypes

mongo_migrations klasörünü kontrol ederseniz, her yerleşik uygulama için bir klasör göreceksiniz. Her klasör kendi geçişlerini içerir.

Tüm bu MongoDB yapılandırmalarını içeren bir Django proje şablonu için şunu çalıştırın:

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

Not: 5.2.x dışında bir Django sürümü kullanıyorsanız, sürümünüzün ilk iki rakamıyla eşleşecek şekilde bu iki numarayı değiştirin.

Uygulamanızı Oluşturun ve Yapılandırın

Bir sonraki adım, ham ve işlenmiş işlem kayıtlarınızı depolamak için bir Django uygulaması oluşturmaktır.

python manage.py startapp sales

Yeni uygulamanızı ObjectId kullanacak şekilde yapılandırmak için sales/apps.py dosyasını açın ve default_auto_field = 'django.db.models.BigAutoField' satırını şu şekilde değiştirin:

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'

Alternatif olarak, yukarıdaki değişikliği içeren şu startapp şablonunu kullanabilirsiniz:

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

Şimdi, pyspark_tutorial/settings.py içinde sales uygulamanızı INSTALLED_APPS listesine ekleyin:

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

Bir MongoDB Dağıtımı Oluşturun

Verilerinizi MongoDB'de depolayacağınız için, verilerinizi bulutta depolayıp yönetmek üzere MongoDB Atlas'ta ücretsiz katmanlı bir dağıtım oluşturma adımlarını anlatacağım.

Atlas hesabı için kaydolun; Google hesabınızı veya bir e-posta adresini kullanabilirsiniz.

Bir Free küme oluşturmak için Create'e tıklayın:

Şimdi açılan sayfada aşağıdaki seçenekleri belirleyin:

• Ücretsiz küme seçeneği
• Küme adı: Cluster0
• Sağlayıcı: AWS
• Bölge: us-east-1

Create Deployment'a tıklayın:

Kullanıcı adınızı ve parolanızı göreceksiniz. Şunları yapın:

1. Kullanıcı adınızı ve parolanızı güvenli bir belgeye kopyalayın.

2. Create Database User'a tıklayın.

3. Choose a connection method'a tıklayarak bir bağlantı IP adresi ayarlayın.

Açılan sayfada şunları yapın:

1. Drivers'ı seçin.

2. Sürücü olarak Python'ı seçin.

3. Install your driver seçeneğindeki komutu kopyalayın ve terminalinizde çalıştırın.

4. Parolanız dahil connection string'inizi, Add your connection string into your application code bölümünün altından kopyalayın ve güvenli bir belgede saklayın.

5. Done'a tıklayın.

Django'yu MongoDB'ye Bağlayın

pyspark_tutorial/settings.py dosyasını açın ve kaydettiğiniz connection string ile Django MongoDB Backend'i kullanmak için DATABASES ayarını güncelleyin. Bir veritabanı adı da belirleyin:

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

Yukarıdaki kodda, <connection string> ifadesini kaydettiğiniz bağlantı dizesi ile değiştirdiğinizden emin olun.

Bu yapılandırma, Django'yu MongoDB Atlas kümenize bağlar. ENGINE MongoDB arka ucunu işaret eder, HOST bağlantı dizenizi saklar ve NAME Django'nun kullanacağı veritabanı adını tanımlar.

Django Uygulamasını Oluşturun

Uygulamanız yapılandırıldığına göre, tarayıcıda işlem verilerini görüntülemek için gerekli modelleri, görünümleri, URL'leri ve şablonları oluşturacaksınız.

Modeli tanımlayın

Modeller, verilerinizin yapısını açıklar. İşlem kayıtlarını temsil eden bir model oluşturun. sales/models.py dosyasını açın ve kodu aşağıdakiyle değiştirin:

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

Bu model, her işlemi ürün ayrıntıları, fiyat, adet ve ülke alanlarıyla tanımlar. İşlemleri order_id'ye göre sıralar. Sorgu performansını artırmak için kilit alanlarda indeksler ve toplam işlem tutarını hesaplayan bir özellik içerir.

Görünümü tanımlayın

Görünümler, bu kayıtların tarayıcıda nasıl görüneceğini kontrol eder. İşlem kayıtlarını görüntülemek için bir görünüm oluşturun. sales/views.py dosyasını açın ve kodu aşağıdakiyle değiştirin:

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

Yukarıdaki görünüm, Transaction modeli ile MongoDB'den tüm işlem kayıtlarını getirir. Tüm işlemler için total_amount alanını toplayarak toplam geliri hesaplar. Ardından, iki değeri daha sonra oluşturacağımız transaction_list.html şablonuna gönderir:

* transactions: bir tabloda görüntülenecek tüm işlem kayıtlarının listesi

* total_revenue: tüm işlemlerden elde edilen toplam tutar

Şablon bu değerleri her bir işlemi ve toplam geliri göstermek için kullanacaktır.

URL'leri yapılandırın

Kullanıcılar bir URL'yi ziyaret ettiğinde Django'nun hangi görünümü yükleyeceğini bilmesi için yönlendirmeler oluşturmanız gerekir. Bir sales/urls.py dosyası oluşturun ve şunları ekleyin:

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

Uygulamanızın URL'lerini proje URL yapılandırmanıza dahil edin. pyspark_tutorial/urls.py dosyasını açın ve aşağıdaki gibi güncelleyin:

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

Şablon oluşturun

Şablonlar, verilerinizin tarayıcıda nasıl görüneceğini tanımlar. Bir sales/templates/sales klasörü oluşturun ve içine transaction_list.html adlı bir dosya ekleyin.

sales uygulama yapınız şu şekilde olmalıdır:

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

transaction_list.html dosyanıza aşağıdakileri ekleyin:

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

Yukarıdaki şablon, görünümden aktarılan verileri dinamik olarak görüntülemek için Django'nun şablon dilini kullanır.

En üstte toplam gelir {{ total_revenue }} kullanılarak gösterilir. Ardından şablon, {% for t in transactions %} etiketiyle transactions içindeki tüm işlem kayıtları arasında döner ve her kaydı tabloda bir satır olarak gösterir. Her sütun bir işlemin order_id, product, price, quantity ve country gibi belirli bir özelliğini gösterir. Kayıt yoksa, {% empty %} etiketi, boş bir tablo yerine “No transactions available” mesajının görünmesini sağlar.

Stil ekleyin

İşlem sayfanızı görsel olarak daha çekici ve okunması daha kolay hale getirmek için özel CSS stilleri ekleyeceksiniz. Django, CSS, JavaScript ve görseller gibi statik dosyaları static adlı özel bir dizin aracılığıyla sunar. Bu, tasarım dosyalarını kodunuzdan ayrı tutar.

Şimdi bir sales/static/sales klasörü oluşturun ve içine styles.css adlı bir dosya ekleyin. sales klasör yapınız şu şekilde görünmelidir:

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 içinde STATIC_URL'ün tanımlı olduğundan emin olun:

STATIC_URL = 'static/'

sales/static/sales/styles.css dosyanıza şunları ekleyin:

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

Son olarak, Django'ya statik dosyaları yüklemesini söyleyin ve CSS dosyasını HTML şablonunuza dahil edin.

sales/templates/sales/transaction_list.html dosyanızın en üstüne {% load static %} ekleyin ve ardından <head> öğesini styles.css dosyasına bağlantı verecek şekilde güncelleyin:

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

Geçişleri çalıştırın

Django'nun MongoDB'de gerekli koleksiyonları oluşturabilmesi için geçişlerinizi oluşturup uygulayın:

python manage.py makemigrations
python manage.py migrate

Verileri MongoDB'ye Yazın

Şimdi, Django etkileşimli kabuğunu kullanarak pyspark_tutorial veritabanınıza uygulamanızın modelleriyle işlem kayıtları ekleyelim:

python manage.py shell

Ardından, satış uygulamanızdan Transaction modelini ve Django yardımcılarından timezone modülünü içe aktarın:

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

Şimdi bulk_create() kullanarak işlem kayıtları ekleyin:

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

Kayıtların başarıyla eklendiğini doğrulayın:

Transaction.objects.count()

Sayı 20 dönerse, kayıtlar başarıyla kaydedilmiştir.

Şimdi exit() komutuyla kabuktan çıkın ve işlemlerinizi tarayıcıda görüntülemek için Django sunucunuzu başlatın:

python manage.py runserver

http://127.0.0.1:8000/ adresini ziyaret ederek verilerinizin görüntülendiğini doğrulayın.

Verileri PySpark ile İşleyin

Kayıtlar artık MongoDB veritabanınızda depolandığına göre, verilerinizi işlemek için Apache Spark kullanacaksınız. Apache Spark, büyük veri kümelerini işlemek için Django projenizde kullanabileceğiniz bir Python API'si, PySpark ile birlikte gelir. PySpark'a yeniyseniz, PySpark kurulum rehberimize göz atın.

PySpark kullanarak MongoDB'ye bağlanan ve işlem kayıtlarınızı okuyan bir Python betiği oluşturacaksınız. Bu veriler üzerinde filtreleme gibi temel işlemler gerçekleştireceksiniz. Ardından ülkeye göre gruplandıracak ve her biri için toplam geliri hesaplayacaksınız. Son olarak, işlenmiş verileri veritabanınızda yeni bir MongoDB koleksiyonuna yazacaksınız.

PySpark'ı kurun

Sunucuyu durdurun ve en güncel MongoDB Spark bağlayıcısı ile uyumlu PySpark sürümünü kurun:

pip install pyspark==3.5.0

Başarılı kurulumunu doğrulayın:

pyspark --version

Spark'a hoş geldiniz mesajı görmelisiniz.

PySpark betik dosyasını oluşturun

Sonraki adımda, proje klasörünüzde, manage.py ile aynı seviyede bir transactions.py dosyası oluşturun. Klasör yapınız artık şöyle görünmelidir:

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

Bağlantı dizenizi oluşturun

PySpark'ı MongoDB'ye bağlamak için, veritabanı adınızı ve koleksiyon adınızı içeren bir bağlantı dizisi oluşturmanız gerekir.  MongoDB'de bir koleksiyon, ilişkisel veritabanlarındaki bir tabloya benzer ve ilgili belgeleri depolar. Koleksiyon adınız, uygulama adınız ve model adınızın alt çizgi (_) ile ayrılmasından oluşur.

Oluşturduğunuz bağlantı dizisini transactions.py dosyanızda kullanın. Doğru biçim şu şekildedir:

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

Yer tutucu değerlerin açıklaması şöyledir:

• <mongodb username> ve <mongodb password> MongoDB Atlas kullanıcı adınız ve parolanızdır.
• <cluster address> Atlas bağlantı dizenizde gösterilen alan adıdır — örneğin cluster0.2rvn82q.mongodb.net.
• <database name> settings.py içinde veritabanınıza verdiğiniz addır.
• <app name>_<model name> Django uygulamanız ve modelinizden üretilen koleksiyonun adıdır — örneğin sales_transaction.

Bu bölümde PySpark kodunuzda kullanacağınız bağlantı dizesi şu şekilde görünecek:

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

PySpark kodunu ekleyin

Aşağıdaki kodu transactions.py dosyanıza ekleyin:

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'ı kullanmanın giriş noktasıdır. Spark ile etkileşim kurmanızı ve veri işlemleri gerçekleştirmenizi sağlar.

spark = SparkSession.builder yeni bir Spark oturumu başlatır ve ona ReadTransactions adını verir. İki .config() seçeneği, Spark'ın MongoDB'ye nasıl bağlanacağını tanımlar:

• spark.mongodb.read.connection.uri: veritabanınızın konumunu Spark'a bildiren MongoDB bağlantı dizesi
• spark.jars.packages: Spark'ın MongoDB ile iletişim kurabilmesi için MongoDB Spark bağlayıcı paketini indirir
• df = spark.read.format('mongodb').load() bağlantı dizenizde tanımlanan koleksiyondaki tüm belgeleri bir PySpark DataFrame'ine yükler; böylece verilerinizi sorgulamak ve dönüştürmek kolaylaşır.
• .show() MongoDB verilerinizi doğrudan terminalde tablo biçiminde önizleme olarak yazdırır.
• spark.stop() işi tamamlandığında Spark uygulamasını sonlandırır ve sistem kaynaklarını serbest bırakır.

Son olarak, <built connection string> ifadesini oluşturduğunuz MongoDB bağlantı dizesi ile değiştirin. Ayrıca, bağlanmadan önce geçerli IP adresinizin MongoDB Atlas ağ erişim listesinde beyaz listede olduğundan emin olun.

Şimdi python transactions.py komutunu çalıştırın; veritabanınızda saklanan işlem kayıtlarını terminalinizde göreceksiniz.

PySpark SQL ile verileri filtreleyin

Kayıtlarınızı filtrelemek için PySpark SQL işlevlerini de kullanabilirsiniz — örneğin yalnızca Nijerya'dan (NG) gelen işlemleri göstermek için. transactions.py dosyanızdaki kodu değiştirin:

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 komutunu tekrar çalıştırın; NG ülkesine ait işlemleri göreceksiniz.

Gruplayın ve toplayın

İşlem kayıtlarınızı gruplandırabilir ve ülke başına toplam geliri hesaplayabilirsiniz. transactions.py dosyanızdaki kodu şu şekilde değiştirin:

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 komutunu çalıştırın. Ülke başına toplam geliri gösteren bir tablo göreceksiniz.

İşlenmiş verileri MongoDB'ye geri yazın

Artık PySpark ile verilerinizi nasıl okuyup işleyeceğinizi anladığınıza göre, işlenmiş verileri bir MongoDB koleksiyonuna nasıl yazacağınızı göstereceğim.

Burada, ülke başına toplam geliri içeren DataFrame'i yeni bir koleksiyona yazacaksınız.

transactions.py dosyanızdaki kodu değiştirin:

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

Yukarıdaki kodda Spark, revenue_per_country adlı toplulaştırılmış DataFrame'i, MongoDB'deki pyspark_tutorial veritabanınızda revenue_per_country adlı yeni bir koleksiyona yazar.

Son olarak, her iki bağlantı yapılandırmasında da '<built connection string>' ifadesini oluşturduğunuz MongoDB bağlantı dizesi ile değiştirdiğinizden emin olun. Ardından python transactions.py komutunu çalıştırın. 

İşlenmiş Verileri Django'da Görüntüleyin

Şimdi, ülke bazında toplam gelirinizi bir Django sayfasında görüntüleyelim.

Önce, ülke başına gelir için yeni bir model oluşturalım. sales/models.py dosyanızın en altına bunu ekleyin:

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

Ardından, sales/views.py dosyanızı revenue_per_country koleksiyonunu getirip görüntüleyecek şekilde değiştirin:

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 içinde yeni görünüme bir rota ekleyin:

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

Sonraki adımda, ülkeye göre toplam geliri görüntülemek için yeni bir şablon oluşturun.

sales/templates/sales klasöründe bir revenue_per_country.html dosyası oluşturun ve şunları ekleyin:

{% 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 şablonunu da ülke başına gelir sayfasına bir bağlantı içerecek şekilde değiştirmeniz gerekir. <h3> etiketinden sonra ancak <table> etiketinden önce aşağıdaki kodu ekleyin:

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

Django sunucusunu çalıştırın:

python manage.py runserver

http://127.0.0.1:8000/revenue_per_country/ adresini ziyaret ederek yeni sayfanın doğru görüntülendiğini doğrulayın.

Tebrikler! Apache Spark'ı kullanarak verilerinizi başarıyla işlediniz, MongoDB'de depoladınız ve işlenmiş verileri Django kullanarak bir web sayfasında görüntülediniz.