一、个人简介

💖💖作者：计算机编程果茶熊
💙💙个人简介：曾长期从事计算机专业培训教学，担任过编程老师，同时本人也热爱上课教学，擅长Java、微信小程序、Python、Golang、安卓Android等多个IT方向。会做一些项目定制化开发、代码讲解、答辩教学、文档编写、也懂一些降重方面的技巧。平常喜欢分享一些自己开发中遇到的问题的解决办法，也喜欢交流技术，大家有技术代码这一块的问题可以问我！
💛💛想说的话：感谢大家的关注与支持！
💜💜
网站实战项目
安卓/小程序实战项目
大数据实战项目
计算机毕业设计选题
💕💕文末获取源码联系计算机编程果茶熊

二、系统介绍

大数据框架：Hadoop+Spark（本次没用Hive，支持定制）
开发语言：Python
后端框架：Django
前端：Vue
详细技术点：Hadoop、HDFS、Spark、Spark SQL、Pandas、NumPy
数据库：MySQL

基于大数据的胆结石消化系统疾病数据分析系统是一个面向医疗健康领域的数据分析平台，该系统采用Hadoop与Spark作为底层大数据处理框架，通过HDFS实现海量医疗数据的分布式存储，利用Spark SQL完成数据的快速查询与计算任务。在技术架构方面，后端使用Django框架搭建服务接口，前端采用Vue构建交互界面，数据库选用MySQL进行结构化数据的管理；在数据处理层面，系统结合Pandas与NumPy进行数据清洗、转换以及统计分析工作。功能模块涵盖了数据大屏展示、用户权限管理、胆结石数据录入与查询，同时提供患者人群特征分析、身体成分分析、血液血脂代谢分析、肝脏功能健康分析以及合并炎症风险分析等多维度的数据挖掘能力，通过这样的方式，医护人员能够从不同角度观察患者的健康状况，系统希望能为胆结石疾病的预防与诊疗提供数据支撑，帮助医疗机构更好地理解患者群体的疾病特征与健康风险分布情况。

三、视频解说

基于大数据的胆结石消化系统疾病数据分析系统–演示视频

四、部分功能展示

五、部分代码展示

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, count, sum, when, round, year, month, desc, asc
from django.http import JsonResponse
from django.views import View
import pandas as pd
import numpy as np
from datetime import datetime
import json
spark = SparkSession.builder.appName("GallstoneAnalysis").config("spark.sql.warehouse.dir", "/user/hive/warehouse").config("spark.executor.memory", "2g").config("spark.driver.memory", "1g").getOrCreate()
class PatientCharacteristicAnalysis(View):
    def post(self, request):
        try:
            request_data = json.loads(request.body)
            start_date = request_data.get('start_date', '2020-01-01')
            end_date = request_data.get('end_date', datetime.now().strftime('%Y-%m-%d'))
            age_group_ranges = [(0, 30), (31, 45), (46, 60), (61, 100)]
            patient_df = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/gallstone_db").option("dbtable", "patient_info").option("user", "root").option("password", "123456").option("driver", "com.mysql.cj.jdbc.Driver").load()
            filtered_patients = patient_df.filter((col("diagnosis_date") >= start_date) & (col("diagnosis_date") <= end_date))
            age_conditions = [when((col("age") >= age_range[0]) & (col("age") <= age_range[1]), f"{age_range[0]}-{age_range[1]}岁") for age_range in age_group_ranges]
            age_column = age_conditions[0]
            for condition in age_conditions[1:]:
                age_column = age_column.otherwise(condition)
            patient_with_age_group = filtered_patients.withColumn("age_group", age_column)
            age_distribution = patient_with_age_group.groupBy("age_group").agg(count("patient_id").alias("patient_count"), round(avg("age"), 2).alias("avg_age")).orderBy("age_group")
            gender_distribution = patient_with_age_group.groupBy("gender").agg(count("patient_id").alias("patient_count")).withColumn("percentage", round(col("patient_count") / filtered_patients.count() * 100, 2))
            region_distribution = patient_with_age_group.groupBy("region").agg(count("patient_id").alias("patient_count")).orderBy(desc("patient_count")).limit(10)
            occupation_risk = patient_with_age_group.groupBy("occupation").agg(count("patient_id").alias("patient_count"), round(avg("stone_size"), 2).alias("avg_stone_size")).orderBy(desc("patient_count"))
            age_result = [{"age_group": row["age_group"], "patient_count": row["patient_count"], "avg_age": float(row["avg_age"])} for row in age_distribution.collect()]
            gender_result = [{"gender": row["gender"], "patient_count": row["patient_count"], "percentage": float(row["percentage"])} for row in gender_distribution.collect()]
            region_result = [{"region": row["region"], "patient_count": row["patient_count"]} for row in region_distribution.collect()]
            occupation_result = [{"occupation": row["occupation"], "patient_count": row["patient_count"], "avg_stone_size": float(row["avg_stone_size"])} for row in occupation_risk.collect()]
            response_data = {"age_distribution": age_result, "gender_distribution": gender_result, "region_distribution": region_result, "occupation_risk": occupation_result, "total_patients": filtered_patients.count()}
            return JsonResponse({"code": 200, "message": "患者人群特征分析完成", "data": response_data})
        except Exception as e:
            return JsonResponse({"code": 500, "message": f"分析过程出现异常: {str(e)}"})
class BloodLipidMetabolismAnalysis(View):
    def post(self, request):
        try:
            request_data = json.loads(request.body)
            patient_ids = request_data.get('patient_ids', [])
            analysis_type = request_data.get('analysis_type', 'all')
            blood_df = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/gallstone_db").option("dbtable", "blood_test_records").option("user", "root").option("password", "123456").option("driver", "com.mysql.cj.jdbc.Driver").load()
            if patient_ids and len(patient_ids) > 0:
                blood_filtered = blood_df.filter(col("patient_id").isin(patient_ids))
            else:
                blood_filtered = blood_df
            cholesterol_levels = blood_filtered.withColumn("cholesterol_level", when(col("total_cholesterol") < 5.2, "正常").when((col("total_cholesterol") >= 5.2) & (col("total_cholesterol") < 6.2), "边缘升高").otherwise("升高"))
            cholesterol_stats = cholesterol_levels.groupBy("cholesterol_level").agg(count("record_id").alias("record_count"), round(avg("total_cholesterol"), 2).alias("avg_cholesterol"), round(avg("ldl_cholesterol"), 2).alias("avg_ldl"), round(avg("hdl_cholesterol"), 2).alias("avg_hdl"))
            triglyceride_analysis = blood_filtered.withColumn("triglyceride_status", when(col("triglyceride") < 1.7, "正常").when((col("triglyceride") >= 1.7) & (col("triglyceride") < 2.3), "轻度升高").when((col("triglyceride") >= 2.3) & (col("triglyceride") < 5.6), "中度升高").otherwise("重度升高"))
            triglyceride_stats = triglyceride_analysis.groupBy("triglyceride_status").agg(count("record_id").alias("record_count"), round(avg("triglyceride"), 2).alias("avg_triglyceride"))
            lipid_ratio_df = blood_filtered.withColumn("tc_hdl_ratio", round(col("total_cholesterol") / col("hdl_cholesterol"), 2)).withColumn("ldl_hdl_ratio", round(col("ldl_cholesterol") / col("hdl_cholesterol"), 2))
            ratio_stats = lipid_ratio_df.agg(round(avg("tc_hdl_ratio"), 2).alias("avg_tc_hdl_ratio"), round(avg("ldl_hdl_ratio"), 2).alias("avg_ldl_hdl_ratio"))
            abnormal_lipid_patients = blood_filtered.filter((col("total_cholesterol") >= 6.2) | (col("triglyceride") >= 2.3) | (col("ldl_cholesterol") >= 4.1) | (col("hdl_cholesterol") < 1.0))
            abnormal_count = abnormal_count_df = abnormal_lipid_patients.select("patient_id").distinct().count()
            correlation_data = blood_filtered.select("total_cholesterol", "triglyceride", "ldl_cholesterol", "hdl_cholesterol").toPandas()
            correlation_matrix = correlation_data.corr().to_dict()
            cholesterol_result = [{"level": row["cholesterol_level"], "count": row["record_count"], "avg_cholesterol": float(row["avg_cholesterol"]), "avg_ldl": float(row["avg_ldl"]), "avg_hdl": float(row["avg_hdl"])} for row in cholesterol_stats.collect()]
            triglyceride_result = [{"status": row["triglyceride_status"], "count": row["record_count"], "avg_triglyceride": float(row["avg_triglyceride"])} for row in triglyceride_stats.collect()]
            ratio_result = ratio_stats.collect()[0]
            response_data = {"cholesterol_analysis": cholesterol_result, "triglyceride_analysis": triglyceride_result, "lipid_ratio": {"avg_tc_hdl_ratio": float(ratio_result["avg_tc_hdl_ratio"]), "avg_ldl_hdl_ratio": float(ratio_result["avg_ldl_hdl_ratio"])}, "abnormal_patient_count": abnormal_count, "correlation_matrix": correlation_matrix, "total_records": blood_filtered.count()}
            return JsonResponse({"code": 200, "message": "血液血脂代谢分析完成", "data": response_data})
        except Exception as e:
            return JsonResponse({"code": 500, "message": f"血脂分析异常: {str(e)}"})
class InflammationRiskAnalysis(View):
    def post(self, request):
        try:
            request_data = json.loads(request.body)
            risk_threshold = request_data.get('risk_threshold', 'medium')
            time_range = request_data.get('time_range', 30)
            inflammation_df = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/gallstone_db").option("dbtable", "inflammation_indicators").option("user", "root").option("password", "123456").option("driver", "com.mysql.cj.jdbc.Driver").load()
            patient_df = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/gallstone_db").option("dbtable", "patient_info").option("user", "root").option("password", "123456").option("driver", "com.mysql.cj.jdbc.Driver").load()
            merged_df = inflammation_df.join(patient_df, inflammation_df.patient_id == patient_df.patient_id, "inner")
            wbc_risk = when(col("wbc_count") > 10, 2).when((col("wbc_count") >= 4) & (col("wbc_count") <= 10), 0).otherwise(1)
            crp_risk = when(col("crp_level") > 10, 3).when((col("crp_level") >= 3) & (col("crp_level") <= 10), 1).otherwise(0)
            neutrophil_risk = when(col("neutrophil_percentage") > 75, 2).when((col("neutrophil_percentage") >= 50) & (col("neutrophil_percentage") <= 75), 0).otherwise(1)
            body_temp_risk = when(col("body_temperature") >= 38.5, 3).when((col("body_temperature") >= 37.5) & (col("body_temperature") < 38.5), 1).otherwise(0)
            risk_scored_df = merged_df.withColumn("wbc_risk_score", wbc_risk).withColumn("crp_risk_score", crp_risk).withColumn("neutrophil_risk_score", neutrophil_risk).withColumn("temp_risk_score", body_temp_risk)
            total_risk_df = risk_scored_df.withColumn("total_risk_score", col("wbc_risk_score") + col("crp_risk_score") + col("neutrophil_risk_score") + col("temp_risk_score"))
            risk_level_df = total_risk_df.withColumn("risk_level", when(col("total_risk_score") >= 8, "高风险").when((col("total_risk_score") >= 4) & (col("total_risk_score") < 8), "中风险").otherwise("低风险"))
            risk_distribution = risk_level_df.groupBy("risk_level").agg(count("inflammation_id").alias("patient_count"), round(avg("total_risk_score"), 2).alias("avg_risk_score"), round(avg("wbc_count"), 2).alias("avg_wbc"), round(avg("crp_level"), 2).alias("avg_crp"))
            high_risk_patients = risk_level_df.filter(col("risk_level") == "高风险").select("patient_id", "patient_name", "age", "gender", "total_risk_score", "wbc_count", "crp_level", "body_temperature").orderBy(desc("total_risk_score")).limit(20)
            age_risk_correlation = risk_level_df.groupBy("age_group").agg(count("inflammation_id").alias("total_count"), sum(when(col("risk_level") == "高风险", 1).otherwise(0)).alias("high_risk_count")).withColumn("high_risk_rate", round(col("high_risk_count") / col("total_count") * 100, 2))
            indicator_stats = risk_level_df.agg(round(avg("wbc_count"), 2).alias("avg_wbc_count"), round(avg("crp_level"), 2).alias("avg_crp_level"), round(avg("neutrophil_percentage"), 2).alias("avg_neutrophil"), round(avg("body_temperature"), 2).alias("avg_temperature"))
            risk_dist_result = [{"risk_level": row["risk_level"], "patient_count": row["patient_count"], "avg_risk_score": float(row["avg_risk_score"]), "avg_wbc": float(row["avg_wbc"]), "avg_crp": float(row["avg_crp"])} for row in risk_distribution.collect()]
            high_risk_result = [{"patient_id": row["patient_id"], "patient_name": row["patient_name"], "age": row["age"], "gender": row["gender"], "risk_score": float(row["total_risk_score"]), "wbc": float(row["wbc_count"]), "crp": float(row["crp_level"]), "temperature": float(row["body_temperature"])} for row in high_risk_patients.collect()]
            age_risk_result = [{"age_group": row["age_group"], "total_count": row["total_count"], "high_risk_count": row["high_risk_count"], "high_risk_rate": float(row["high_risk_rate"])} for row in age_risk_correlation.collect()]
            indicator_result = indicator_stats.collect()[0]
            response_data = {"risk_

六、部分文档展示

七、END

💕💕文末获取源码联系计算机编程果茶熊