# Hadoop+Spark+Django技术栈在公共交通数据分析系统的构建

随着城市化进程加速和智慧交通发展，公共交通系统产生了海量运营数据。基于Hadoop生态体系，结合Spark计算框架和Django Web应用，构建高效的公共交通数据分析系统，对于优化交通资源配置、提升运营效率具有重要意义。本文将深入探讨这一系统的技术实现与应用方案。

## 一、系统架构设计与技术选型

系统采用分层微服务架构，充分发挥各技术组件的优势：

```

数据源层(公交GPS、IC卡、调度日志)

→ 数据湖层(HDFS+Hive)

→ 计算引擎层(Spark+MapReduce)

→ 服务层(Django REST API)

→ 展示层(可视化大屏+移动端)

```

核心技术组件包括：

- Hadoop 3.x分布式存储与计算框架

- Hive 3.x数据仓库与SQL引擎

- Spark 3.x内存计算与流处理

- Django 4.x Web应用框架

- ECharts+DataV可视化库

## 二、基于Hive的多源数据集成

公共交通数据具有多源异构特点，需要通过Hive进行统一管理和集成：

```sql

-- Hive数据仓库表结构设计

-- 公交GPS轨迹数据表

CREATE EXTERNAL TABLE IF NOT EXISTS bus_gps_trajectory (

vehicle_id STRING COMMENT '车辆ID',

line_id STRING COMMENT '线路ID',

gps_time TIMESTAMP COMMENT 'GPS时间',

longitude DECIMAL(10,6) COMMENT '经度',

latitude DECIMAL(10,6) COMMENT '纬度',

speed DECIMAL(5,2) COMMENT '瞬时速度(km/h)',

direction SMALLINT COMMENT '行驶方向',

load_factor DECIMAL(3,2) COMMENT '满载率'

) PARTITIONED BY (dt STRING, hour STRING)

ROW FORMAT DELIMITED

FIELDS TERMINATED BY ','

STORED AS ORC

LOCATION '/hive/warehouse/bus_gps_trajectory'

TBLPROPERTIES ('orc.compress'='SNAPPY');

-- IC卡交易数据表

CREATE EXTERNAL TABLE IF NOT EXISTS ic_card_transaction (

card_id STRING COMMENT '卡号',

trans_time TIMESTAMP COMMENT '交易时间',

vehicle_id STRING COMMENT '车辆ID',

line_id STRING COMMENT '线路ID',

station_id STRING COMMENT '站点ID',

trans_type STRING COMMENT '交易类型(上车/下车)',

amount DECIMAL(6,2) COMMENT '交易金额'

) PARTITIONED BY (dt STRING)

ROW FORMAT DELIMITED

FIELDS TERMINATED BY '|'

STORED AS PARQUET

LOCATION '/hive/warehouse/ic_card_transaction';

-- 线路站点信息维表

CREATE EXTERNAL TABLE IF NOT EXISTS bus_line_station (

line_id STRING COMMENT '线路ID',

station_id STRING COMMENT '站点ID',

station_name STRING COMMENT '站点名称',

station_sequence INT COMMENT '站点序号',

longitude DECIMAL(10,6) COMMENT '经度',

latitude DECIMAL(10,6) COMMENT '纬度',

district STRING COMMENT '所属行政区'

) STORED AS PARQUET

LOCATION '/hive/warehouse/bus_line_station';

```

## 三、Spark数据处理与分析计算

利用Spark进行大规模公共交通数据的处理与分析：

```scala

// Spark SQL公交运营分析

import org.apache.spark.sql.SparkSession

import org.apache.spark.sql.functions._

import org.apache.spark.sql.expressions.Window

object PublicTransportAnalysis {

def main(args: Array[String]): Unit = {

val spark = SparkSession.builder()

.appName("PublicTransportAnalytics")

.config("spark.sql.adaptive.enabled", "true")

.config("spark.sql.adaptive.coalescePartitions.enabled", "true")

.enableHiveSupport()

.getOrCreate()

// 1. 线路运行准点率分析

analyzePunctualityRate(spark)

// 2. 客流时空分布分析

analyzePassengerFlow(spark)

// 3. 站点换乘分析

analyzeTransferPattern(spark)

spark.stop()

}

def analyzePunctualityRate(spark: SparkSession): Unit = {

import spark.implicits._

// 从Hive读取GPS轨迹数据

val gpsDF = spark.sql("""

SELECT

vehicle_id,

line_id,

dt,

hour,

gps_time,

longitude,

latitude,

station_id

FROM bus_gps_trajectory

WHERE dt = '2024-01-15'

""")

// 读取线路时刻表

val scheduleDF = spark.sql("""

SELECT

line_id,

station_id,

scheduled_time

FROM bus_schedule

""")

// 关联计算准点率

val punctualityDF = gpsDF.alias("g")

.join(scheduleDF.alias("s"),

$"g.line_id" === $"s.line_id" &&

$"g.station_id" === $"s.station_id")

.withColumn("time_diff",

abs(unix_timestamp($"g.gps_time") -

unix_timestamp($"s.scheduled_time")))

.withColumn("is_punctual",

when($"time_diff" < 300, 1).otherwise(0))

// 按线路聚合准点率

val resultDF = punctualityDF

.groupBy($"g.line_id", $"g.dt")

.agg(

count("*").as("total_trips"),

sum($"is_punctual").as("punctual_trips"),

(sum($"is_punctual") / count("*") * 100)

.as("punctuality_rate")

)

.orderBy($"punctuality_rate".desc)

// 结果写回Hive

resultDF.write

.mode("overwrite")

.saveAsTable("bus_punctuality_analysis")

}

def analyzePassengerFlow(spark: SparkSession): Unit = {

// 基于IC卡数据的客流分析

val passengerFlowDF = spark.sql("""

WITH boarding_data AS (

SELECT

card_id,

vehicle_id,

line_id,

station_id as boarding_station,

trans_time as boarding_time

FROM ic_card_transaction

WHERE trans_type = '上车'

AND dt = '2024-01-15'

),

alighting_data AS (

SELECT

card_id,

vehicle_id,

station_id as alighting_station,

trans_time as alighting_time

FROM ic_card_transaction

WHERE trans_type = '下车'

AND dt = '2024-01-15'

)

SELECT

b.line_id,

b.boarding_station,

a.alighting_station,

COUNT(DISTINCT b.card_id) as passenger_count,

HOUR(b.boarding_time) as time_hour

FROM boarding_data b

JOIN alighting_data a ON b.card_id = a.card_id

AND b.vehicle_id = a.vehicle_id

GROUP BY b.line_id, b.boarding_station,

a.alighting_station, HOUR(b.boarding_time)

ORDER BY passenger_count DESC

""")

passengerFlowDF.write

.mode("overwrite")

.saveAsTable("passenger_flow_analysis")

}

```

## 四、复杂交通指标计算

实现关键交通性能指标的分布式计算：

```python

# Python/Spark混合计算模式

from pyspark.sql import SparkSession

from pyspark.sql.window import Window

import pyspark.sql.functions as F

from pyspark.sql.types import *

class TransportMetricsCalculator:

def __init__(self):

self.spark = SparkSession.builder \

.appName("TransportMetrics") \

.enableHiveSupport() \

.getOrCreate()

def calculate_line_load_factor(self, date_str):

"""计算线路满载率"""

# 读取GPS数据和IC卡数据

gps_df = self.spark.sql(f"""

SELECT vehicle_id, line_id, gps_time,

longitude, latitude, load_factor

FROM bus_gps_trajectory

WHERE dt = '{date_str}'

""")

# 计算每辆车的实时载客量

passenger_count_df = self.spark.sql(f"""

SELECT vehicle_id,

COUNT(DISTINCT card_id) as real_passenger_count

FROM ic_card_transaction

WHERE dt = '{date_str}'

GROUP BY vehicle_id

""")

# 关联计算满载率

result_df = gps_df.join(

passenger_count_df,

>

how="left"

).withColumn(

"calculated_load_factor",

F.when(

F.col("real_passenger_count").isNotNull(),

F.col("real_passenger_count") / 80 # 假设车辆额定载客80人

).otherwise(F.col("load_factor"))

)

# 按线路时段聚合

line_load_df = result_df.groupBy(

"line_id",

F.hour("gps_time").alias("time_hour")

).agg(

F.avg("calculated_load_factor").alias("avg_load_factor"),

F.max("calculated_load_factor").alias("max_load_factor"),

F.stddev("calculated_load_factor").alias("load_factor_std")

)

return line_load_df

def analyze_transfer_pattern(self):

"""分析乘客换乘模式"""

# 定义UDF处理换乘逻辑

def detect_transfer(card_records):

# 实现换乘检测逻辑

transfers = []

records = sorted(card_records, key=lambda x: x['time'])

for i in range(len(records)-1):

if records[i]['type'] == '下车' and \

records[i+1]['type'] == '上车':

time_diff = (records[i+1]['time'] -

records[i]['time']).total_seconds()

if time_diff < 1800: # 30分钟内换乘

transfers.append({

'from_line': records[i]['line'],

'to_line': records[i+1]['line'],

'transfer_time': time_diff

})

return transfers

detect_transfer_udf = F.udf(

detect_transfer,

ArrayType(MapType(StringType(), StringType()))

)

# 执行换乘分析

transfer_df = self.spark.sql("""

SELECT card_id,

COLLECT_LIST(

STRUCT(

trans_time as time,

trans_type as type,

line_id as line,

station_id as station

)

) as journey_records

FROM ic_card_transaction

WHERE dt = '2024-01-15'

GROUP BY card_id

HAVING COUNT(*) > 1

""")

analyzed_df = transfer_df.withColumn(

"transfers",

detect_transfer_udf("journey_records")

)

return analyzed_df

```

## 五、Django REST API服务设计

构建面向交通管理的数据服务接口：

```python

# Django交通数据API视图

from django.http import JsonResponse

from rest_framework.views import APIView

from rest_framework.response import Response

from django.db import connection

from pyhive import hive

import pandas as pd

import json

class TrafficDataAPI(APIView):

"""交通数据查询API"""

<"r9.j9k5.org.cn"><"k2.j9k5.org.cn"><"p5.j9k5.org.cn">

def get_hive_connection(self):

"""获取Hive连接"""

return hive.Connection(

host='hive-server',

port=10000,

username='hadoop',

database='transport_db'

)

def get_line_performance(self, request):

"""获取线路运行绩效数据"""

line_id = request.GET.get('line_id')

date_str = request.GET.get('date', '2024-01-15')

conn = self.get_hive_connection()

# 执行Hive查询

query = f"""

SELECT

line_id,

dt as analysis_date,

AVG(punctuality_rate) as avg_punctuality,

AVG(avg_load_factor) as avg_load_factor,

SUM(passenger_count) as total_passengers,

COUNT(DISTINCT vehicle_id) as vehicle_count

FROM bus_line_daily_stats

WHERE dt = '{date_str}'

{'AND line_id = ' + line_id if line_id else ''}

GROUP BY line_id, dt

ORDER BY total_passengers DESC

"""

df = pd.read_sql(query, conn)

# 转换为JSON响应

result = df.to_dict('records')

return Response({

'status': 'success',

'count': len(result),

'data': result

})

def get_real_time_traffic(self, request):

"""获取实时交通状况"""

conn = self.get_hive_connection()

# 查询最近15分钟的GPS数据

query = """

SELECT

g.line_id,

g.vehicle_id,

g.longitude,

g.latitude,

g.speed,

g.gps_time,

l.vehicle_count,

l.avg_speed

FROM (

SELECT *,

ROW_NUMBER() OVER (

PARTITION BY vehicle_id

ORDER BY gps_time DESC

) as rn

FROM bus_gps_trajectory_realtime

) g

JOIN bus_line_realtime_stats l

ON g.line_id = l.line_id

WHERE g.rn = 1

ORDER BY g.gps_time DESC

LIMIT 1000

"""

df = pd.read_sql(query, conn)

# 生成GeoJSON格式

geojson = {

"type": "FeatureCollection",

"features": []

}

for _, row in df.iterrows():

feature = {

"type": "Feature",

"geometry": {

"type": "Point",

"coordinates": [row['longitude'], row['latitude']]

},

"properties": {

"line_id": row['line_id'],

"vehicle_id": row['vehicle_id'],

"speed": row['speed'],

"time": row['gps_time'].isoformat()

}

geojson["features"].append(feature)

return Response(geojson)

class StationAnalysisAPI(APIView):

"""站点分析API"""

def get_station_flow(self, request, station_id):

"""获取站点客流分析"""

hour = request.GET.get('hour', None)

date_str = request.GET.get('date')

conn = hive.Connection(host='hive-server', port=10000)

# 构建查询条件

conditions = [f"dt = '{date_str}'", f"station_id = '{station_id}'"]

if hour:

conditions.append(f"time_hour = {hour}")

where_clause = " AND ".join(conditions)

query = f"""

SELECT

time_hour,

SUM(boarding_count) as boarding_count,

SUM(alighting_count) as alighting_count,

AVG(waiting_time) as avg_waiting_time

FROM station_hourly_stats

WHERE {where_clause}

GROUP BY time_hour

ORDER BY time_hour

"""

df = pd.read_sql(query, conn)

# 计算客流转换率

if not df.empty:

df['flow_rate'] = df['boarding_count'] / df['alighting_count'].replace(0, 1)

return Response({

'station_id': station_id,

'date': date_str,

'flow_data': df.to_dict('records')

})

```

## 六、可视化大屏实现

基于ECharts的交通数据可视化：

```javascript

// 交通数据可视化大屏

class TransportDashboard {

constructor() {

this.initDashboard();

this.setupRealTimeUpdates();

this.loadInitialData();

}

initDashboard() {

// 初始化所有图表

this.charts = {

// 线路运行热力图

lineHeatmap: echarts.init(

document.getElementById('line-heatmap')

),

// 客流时序图

passengerFlow: echarts.init(

document.getElementById('passenger-flow')

),

// 站点流量图

stationFlow: echarts.init(

document.getElementById('station-flow')

),

// 实时车辆监控地图

realtimeMap: echarts.init(

document.getElementById('realtime-map')

)

};

// 设置地图选项

this.setupRealtimeMap();

}

setupRealtimeMap() {

const mapOption = {

title: {

text: '实时公交车辆监控',

left: 'center'

},

tooltip: {

formatter: function(params) {

return `

${params.data.line}

</div>

车辆: ${params.data.vehicle}<br/>

速度: ${params.data.speed} km/h<br/>

时间: ${params.data.time}

`;

}

},

geo: {

map: 'city',

roam: true,

zoom: 12,

center: [116.4, 39.9],

itemStyle: {

areaColor: '#323c48',

borderColor: '#111'

},

emphasis: {

itemStyle: {

areaColor: '#2a333d'

}

},

series: [{

type: 'scatter',

coordinateSystem: 'geo',

symbol: 'circle',

symbolSize: function(val) {

return Math.max(8, Math.min(val[2] / 5, 20));

},

data: [],

itemStyle: {

color: function(params) {

// 根据速度着色

const speed = params.data[2];

if (speed < 10) return '#ff0000';

if (speed < 20) return '#ffa500';

if (speed < 40) return '#ffff00';

return '#00ff00';

}

}]

};

this.charts.realtimeMap.setOption(mapOption);

}

async loadInitialData() {

// 加载线路绩效数据

const lineData = await this.fetchLinePerformance();

this.updateLineHeatmap(lineData);

// 加载实时车辆数据

await this.updateRealtimeVehicles();

<"m1.j9k5.org.cn"><"a8.j9k5.org.cn"><"d0.j9k5.org.cn">

}

async fetchLinePerformance() {

const response = await fetch(

'/api/transport/line_performance/?date=2024-01-15'

);

return await response.json();

}

updateLineHeatmap(data) {

const heatmapData = data.map(item => [

item.line_id,

item.time_hour,

item.avg_load_factor

]);

const option = {

tooltip: {

position: 'top'

},

grid: {

height: '50%',

top: '10%'

},

xAxis: {

type: 'category',

data: Array.from({length: 24}, (_, i) => i),

splitArea: {

show: true

}

},

yAxis: {

type: 'category',

data: data.map(item => item.line_id),

splitArea: {

show: true

}

},

visualMap: {

min: 0,

max: 1,

calculable: true,

orient: 'horizontal',

left: 'center',

bottom: '15%'

},

series: [{

name: '满载率',

type: 'heatmap',

data: heatmapData,

label: {

show: true,

formatter: function(params) {

return (params.value[2] * 100).toFixed(0) + '%';

}

},

emphasis: {

itemStyle: {

shadowBlur: 10,

shadowColor: 'rgba(0, 0, 0, 0.5)'

}

}]

};

this.charts.lineHeatmap.setOption(option);

}

setupRealTimeUpdates() {

// 建立WebSocket连接

this.socket = new WebSocket('ws://localhost:8000/ws/traffic/');

this.socket.=> {

const data = JSON.parse(event.data);

if (data.type === 'vehicle_update') {

this.updateVehiclePosition(data.vehicles);

} else if (data.type === 'flow_update') {

this.updatePassengerFlow(data.flow_data);

}

};

// 定时刷新

setInterval(() => {

this.updateRealtimeVehicles();

}, 30000); // 每30秒更新一次

}

```

## 七、系统部署与优化

```yaml

# Docker Compose集群部署

version: '3.8'

services:

namenode:

image: bde2020/hadoop-namenode:2.0.0-hadoop3.2.1-java8

ports:

- "9870:9870"

volumes:

- hadoop_namenode:/hadoop/dfs/name

environment:

- CLUSTER_NAME=hadoop-cluster

hive-server:

image: apache/hive:3.1.3

ports:

- "10000:10000"

depends_on:

- namenode

environment:

- HIVE_SITE_CONF_javax_jdo_option_ConnectionURL=jdbc:mysql://mysql:3306/hive

- HIVE_SITE_CONF_javax_jdo_option_ConnectionDriverName=com.mysql.jdbc.Driver

spark-master:

image: bitnami/spark:3.3.0

ports:

- "8080:8080"

- "7077:7077"

depends_on:

- namenode

environment:

- SPARK_MODE=master

django-app:

build: ./webapp

ports:

- "8000:8000"

depends_on:

- hive-server

- spark-master

volumes:

- ./webapp:/app

- ./logs:/var/log/django

environment:

- HIVE_SERVER=hive-server

- SPARK_MASTER=spark://spark-master:7077

volumes:

hadoop_namenode:

```

## 八、应用价值与展望

该公共交通数据分析系统通过整合Hadoop大数据生态、Spark分布式计算和Django Web应用，实现了对海量交通数据的全面分析和智能应用。系统能够：

1. 实时监控公交运行状态，优化车辆调度

2. 深入分析客流规律，合理配置运力资源

3. 评估线路运营绩效，提升服务质量

4. 预测交通需求变化，辅助规划决策

系统的技术架构具有良好的扩展性，未来可以集成更多数据源（如地铁、出租车、共享单车等），构建城市综合交通大数据平台。同时，通过引入机器学习算法，可以实现更精准的客流预测、异常检测和智能调度，推动公共交通向智慧化、智能化方向发展。

这种基于Hadoop+Spark+Django的技术方案，不仅适用于公共交通领域，也为其他城市基础设施的数字化管理和智慧城市建设提供了可复用的技术框架，具有广泛的应用前景和社会价值。

ITPub博客

Hadoop+Spark+Django技术栈在公共交通数据分析系统的构建

最新文章