from pyspark import SparkConf, SparkContext
conf = SparkConf()
conf.setAppName("My Spark App")
conf.set("spark.executor.memory", "4g")  # 設(shè)置執(zhí)行器內(nèi)存為4GB
sc = SparkContext(conf=conf)

# 在spark-defaults.conf文件中添加以下行
spark.executor.memory 4g

spark-submit --conf "spark.executor.memory=4g" your_spark_app.py

from pyspark.sql import SparkSession
# 創(chuàng)建SparkSession
spark = SparkSession.builder \
    .appName("BankFraudDetection") \
    .config("spark.executor.memory", "4g") \
    .getOrCreate()
# 加載數(shù)據(jù)
bank_transactions = spark.read.format("csv").option("header", "true").load("path/to/bank_transactions.csv")

# 查看數(shù)據(jù)結(jié)構(gòu)
bank_transactions.printSchema()
# 顯示數(shù)據(jù)的前幾行
bank_transactions.show()
# 數(shù)據(jù)清洗，例如：去除非法或缺失的交易記錄
cleaned_transactions = bank_transactions.filter("amount IS NOT NULL AND transaction_date IS NOT NULL")

from pyspark.sql.functions import unix_timestamp, to_date, datediff
# 轉(zhuǎn)換日期格式，并創(chuàng)建新特征
cleaned_transactions = cleaned_transactions.withColumn("transaction_time", unix_timestamp(col("transaction_date"), "yyyy-MM-dd HH:mm:ss"))
    .withColumn("is_weekend", (datediff(to_date("transaction_date"), to_date("transaction_time")) % 7) >= 5)

# 選擇相關(guān)特征列
selected_features = cleaned_transactions.select("account_id", "transaction_time", "amount", "is_weekend")

from pyspark.ml.classification import LogisticRegression
# 將數(shù)據(jù)集分為訓(xùn)練集和測試集
train_data, test_data = selected_features.randomSplit([0.8, 0.2])
# 轉(zhuǎn)換數(shù)據(jù)為二分類問題，假設(shè)1為欺詐交易，0為正常交易
labeled_data = train_data.withColumn("label", when(train_data["is_fraud"], 1).otherwise(0))
# 創(chuàng)建邏輯回歸模型
lr = LogisticRegression(featuresCol="features", labelCol="label")
# 訓(xùn)練模型
model = lr.fit(labeled_data)

# 使用測試集進(jìn)行預(yù)測
predictions = model.transform(test_data)
# 評估模型
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="label")
auc = evaluator.evaluate(predictions)
print(f"Area Under the ROC Curve (AUC) = {auc:.2f}")

# 將模型保存到磁盤
model.save("path/to/model")
# 加載模型進(jìn)行預(yù)測
loaded_model = LogisticRegressionModel.load("path/to/model")
# 對新數(shù)據(jù)進(jìn)行預(yù)測
new_transactions = spark.createDataFrame([...])  # 新的交易數(shù)據(jù)
predictions_new = loaded_model.transform(new_transactions)

from pyspark.sql import SparkSession
def create_spark_session():
    spark = SparkSession.builder \
        .appName("DjangoSparkIntegration") \
        .config("spark.executor.memory", "4g") \
        .getOrCreate()
    return spark

# 假設(shè)這是你的數(shù)據(jù)分析函數(shù)
def perform_data_analysis(spark, data_path):
    df = spark.read.csv(data_path, header=True, inferSchema=True)
    # 數(shù)據(jù)清洗、特征工程等操作...
    return df  # 或者返回模型、結(jié)果等

from django.db import models
class AnalysisResult(models.Model):
    result_value = models.FloatField()
    created_at = models.DateTimeField(auto_now_add=True)
def save_results_to_db(results, model_class):
    for result in results:
        model_class.objects.create(result_value=result)

from django.http import JsonResponse
from django.views import View
class數(shù)據(jù)分析結(jié)果View(View):
    def get(self, request, *args, **kwargs):
        spark = create_spark_session()
        results_df = perform_data_analysis(spark, 'path/to/your/data')
        # 假設(shè)results_df已經(jīng)是可以迭代的結(jié)果集
        results_list = results_df.collect()  # 或使用其他方法轉(zhuǎn)換結(jié)果
        save_results_to_db(results_list, AnalysisResult)
        # 構(gòu)建響應(yīng)數(shù)據(jù)
        response_data = {
            'status': 'success',
            'results': [(row['result_value'], row['created_at']) for row in results_list]
        }
        return JsonResponse(response_data)

from rest_framework import serializers, viewsets
class AnalysisResultSerializer(serializers.ModelSerializer):
    class Meta:
        model = AnalysisResult
        fields = ['id', 'result_value', 'created_at']
class AnalysisResultViewSet(viewsets.ModelViewSet):
    queryset = AnalysisResult.objects.all()
    serializer_class = AnalysisResultSerializer

from django.urls import path
from .views import 數(shù)據(jù)分析結(jié)果View
from rest_framework.routers import DefaultRouter
from .views import AnalysisResultViewSet
router = DefaultRouter()
router.register(r'analysis_results', AnalysisResultViewSet)
urlpatterns = [
    path('data_analysis/', 數(shù)據(jù)分析結(jié)果View.as_view(), name='data_analysis'),
] + router.urls

<!-- example.html -->
{% extends 'base.html' %}
{% block content %}
  <h1>數(shù)據(jù)分析結(jié)果</h1>
  <ul>
    {% for result in results %}
      <li>結(jié)果值: {{ result.result_value }} - 時間: {{ result.created_at }}</li>
    {% endfor %}
  </ul>
{% endblock %}

# 使用django-cron
from django_cron import CronJobBase, Schedule
class ScheduledAnalysisJob(CronJobBase):
    schedule = Schedule(run_every_mins=60)  # 每小時執(zhí)行一次
    code = 'myapp.cron.run_analysis'
    def do(self):
        spark = create_spark_session()
        perform_data_analysis(spark, 'path/to/your/data_regular')