使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析

简介: 使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析

使用spark基于出租车GPS数据实现车辆数量统计以及北京每个城区的车辆位置点数分析

本文将介绍如何使用pyspark以及scala实现的spark分析出租车GPS数据,具体来说,我们将计算每个北京城区内的车辆位置点数,以及统计出租车的数量。我们将使用两个数据集:district.txt 包含北京各城区的中心坐标和半径,taxi_gps.txt 包含出租车的GPS位置数据。以下是数据文件的示例内容

1、数据解析

出租车GPS数据文件(taxi_gps.txt)

北京区域中心坐标及半径数据文件(district.txt)

2、需求分析

能够输出以下统计信息

·A:该出租车GPS数据文件(taxi_gps.txt)包含多少量车?

·B:北京每个城区的车辆位置点数(每辆车有多个位置点,允许重复)


A输出:

·以第一列统计车辆数,去重·输出


B输出:

1.从(district.txt)文件中取第一个区的记录,获得其名称D、中心坐标M(xo,yo)和半径r;

2.从(taxi_gps.txt)中获取第一条位置点记录,获得其坐标N(xp,y)

3.利用欧几里得距离计算公式计算点M和N的距离dis,如果dis

4.继续从2开始循环,获得第二个位置记录;直至所有记录遍历完。·5.继续从1开始循环,获得第二个区的记录 district.txt

3、统计出租车的数量

接下来,我们统计出租车的数量。我们可以简单地读取taxi_gps.txt文件,然后使用countDistinct函数来统计不同车辆标识的数量。

python实现该功能的代码

from pyspark.sql import SparkSession
from pyspark.sql.functions import countDistinct

# 创建一个SparkSession
spark = SparkSession.builder \
    .getOrCreate()

# 读取出租车GPS数据
taxi_df = spark.read.csv("data/taxi_gps.txt", header=False, inferSchema=True)

# 计算唯一出租车的数量
num_taxis = taxi_df.select(countDistinct("_c0")).collect()[0][0]

# 输出结果
print("出租车的数量为:", num_taxis)

# 停止SparkSession
spark.stop()

scala实现该功能的代码

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.countDistinct

object CarCount{
  def main(args: Array[String]): Unit = {
    // 创建一个SparkSession
    val spark = SparkSession.builder()
      .appName("TaxiGPS")
      .master("local")
      .getOrCreate()

    // 读取出租车GPS数据
    val taxiDF = spark.read
      .option("header", "false")
      .option("inferSchema", "true")
      .csv("data/taxi_gps.txt")

    // 计算唯一出租车的数量
    val numTaxis = taxiDF.select(countDistinct("_c0")).collect()(0)(0)

    // 输出结果
    println(s"出租车的数量为: $numTaxis")

    // 停止SparkSession
    spark.stop()

  }

}

4、计算每个城区内每辆车的位置点数

首先,我们使用PySpark读取数据并计算每个城区内每辆车的位置点数。为了实现这一点,我们需要计算每个出租车位置与各城区中心的距离,然后检查距离是否在城区的半径范围内。

python实现该功能的代码

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, countDistinct, udf

# 创建一个SparkSession
spark = SparkSession.builder \
    .getOrCreate()

# 读取区域信息和出租车GPS数据
district_df = spark.read.csv("data/district.txt", header=False, inferSchema=True)
taxi_df = spark.read.csv("data/taxi_gps.txt", header=False, inferSchema=True)

# 提取区域信息
district_info = district_df.select(col("_c0").alias("area"),
                                   col("_c1").cast("double").alias("center_a"),
                                   col("_c2").cast("double").alias("center_b"),
                                   col("_c3").cast("double").alias("radio"))

# 定义UDF以计算两点之间的欧几里得距离
def euclidean_distance(x1, y1, x2, y2):
    return ((x1 - x2) ** 2 + (y1 - y2) ** 2) ** 0.5

calculate_distance = udf(euclidean_distance)

# 计算每个城区内每辆车的位置点数
result_df = district_info.crossJoin(taxi_df) \
    .withColumn("distance", calculate_distance(col("center_a"), col("center_b"), col("_c4"), col("_c5"))).createTempView("car")


spark.sql("select _c0 as car,count(distinct(area)) as cnt  from car where distance*1000 < radio group by _c0").show()


spark.stop()

scala实现该功能的代码

package org.example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.UserDefinedFunction

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

    // 创建一个SparkSession
    val spark = SparkSession.builder()
      .appName("TaxiGPS")
      .master("local[*]")
      .getOrCreate()

    // 读取区域信息和出租车GPS数据
    val districtDF = spark.read
      .option("header", "false")
      .option("inferSchema", "true")
      .csv("data/district.txt")

    val taxiDF = spark.read
      .option("header", "false")
      .option("inferSchema", "true")
      .csv("data/taxi_gps.txt")

    // 提取区域信息
    val districtInfo = districtDF.select(
      col("_c0").alias("area"),
      col("_c1").cast("double").alias("center_a"),
      col("_c2").cast("double").alias("center_b"),
      col("_c3").cast("double").alias("radio")
    )

    // 定义UDF以计算两点之间的欧几里得距离
    def euclideanDistance(x1: Double, y1: Double, x2: Double, y2: Double): Double = {
      math.sqrt(math.pow(x1 - x2, 2) + math.pow(y1 - y2, 2))
    }

    val calculateDistance: UserDefinedFunction = udf(euclideanDistance _)

    // 计算每个城区内每辆车的位置点数
    val resultDF = districtInfo.crossJoin(taxiDF)
      .withColumn("distance", calculateDistance(col("center_a"), col("center_b"), col("_c4"), col("_c5")))

    resultDF.createOrReplaceTempView("car")

    spark.sql("SELECT _c0 AS car, COUNT(DISTINCT area) AS cnt FROM car WHERE distance * 1000 < radio GROUP BY _c0").show()

    // 停止SparkSession
    spark.stop()
  }

}

总结

通过以上两个代码示例,我们使用PySpark成功地计算了北京各城区内每辆车的位置点数,并统计了出租车的数量。这些分析可以帮助我们更好地理解出租车在各个城区的分布情况,进而为城市交通管理提供数据支持。PySpark强大的数据处理能力和灵活的编程接口,使得我们能够轻松地处理和分析大规模的GPS数据。




相关文章
|
13天前
|
机器学习/深度学习 分布式计算 算法
Spark快速大数据分析PDF下载读书分享推荐
《Spark快速大数据分析》适合初学者,聚焦Spark实用技巧,同时深入核心概念。作者团队来自Databricks,书中详述Spark 3.0新特性,结合机器学习展示大数据分析。Spark是大数据分析的首选工具,本书助你驾驭这一利器。[PDF下载链接][1]。 ![Spark Book Cover][2] [1]: https://zhangfeidezhu.com/?p=345 [2]: https://i-blog.csdnimg.cn/direct/6b851489ad1944548602766ea9d62136.png#pic_center
39 1
Spark快速大数据分析PDF下载读书分享推荐
|
2月前
|
移动开发 分布式计算 Spark
Spark的几种去重的原理分析
Spark的几种去重的原理分析
27 0
|
13天前
|
弹性计算 分布式计算 Serverless
全托管一站式大规模数据处理和分析Serverless平台 | EMR Serverless Spark 评测
【7月更文挑战第6天】全托管一站式大规模数据处理和分析Serverless平台 | EMR Serverless Spark 评测
23580 42
|
3天前
|
分布式计算 数据处理 流计算
实时计算 Flink版产品使用问题之使用Spark ThriftServer查询同步到Hudi的数据时,如何实时查看数据变化
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
19天前
|
弹性计算 分布式计算 DataWorks
DataWorks产品使用合集之spark任务如何跨空间取表数据
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
12 1
|
1月前
|
SQL 分布式计算 HIVE
实时计算 Flink版产品使用问题之同步到Hudi的数据是否可以被Hive或Spark直接读取
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
1月前
|
SQL 分布式计算 NoSQL
使用Spark高效将数据从Hive写入Redis (功能最全)
使用Spark高效将数据从Hive写入Redis (功能最全)
|
1月前
|
机器学习/深度学习 数据采集 分布式计算
基于spark的大数据分析预测地震受灾情况的系统设计
基于spark的大数据分析预测地震受灾情况的系统设计
|
2月前
|
SQL 分布式计算 关系型数据库
Spark 分析计算连续三周登录的用户数
本文介绍了如何使用窗口函数`range between`来查询`login_time`为2022-03-10的用户最近连续三周的登录数。首先在MySQL中创建`log_data`表并插入数据,接着定义需求为找出该日期前连续三周活跃的用户数。通过Spark SQL,分步骤实现:1)确定统计周期,2)筛选符合条件的数据,3)计算用户连续登录状态。在初始实现中出现错误,因未考虑日期在周中的位置,修正后正确计算出活跃用户数。
|
2月前
|
SQL 分布式计算 关系型数据库
使用 Spark 抽取 MySQL 数据到 Hive 时某列字段值出现异常(字段错位)
在 MySQL 的 `order_info` 表中,包含 `order_id` 等5个字段,主要存储订单信息。执行按 `create_time` 降序的查询,显示了部分结果。在 Hive 中复制此表结构时,所有字段除 `order_id` 外设为 `string` 类型,并添加了 `etl_date` 分区字段。然而,由于使用逗号作为字段分隔符,当 `address` 字段含逗号时,数据写入 Hive 出现错位,导致 `create_time` 值变为中文字符串。问题解决方法包括更换字段分隔符或使用 Hive 默认分隔符 `\u0001`。此案例提醒在建表时需谨慎选择字段分隔符。