利用Spark进行房地产分析 #Hadoop Spark Mysql_基于hadoop房产大数据的可视化分析

文章目录

---

前言

   本文对如何开发基于spark和Hadoop的大数据分析平台进行了广泛和深入的研究，其范围包括python爬虫、Java、spark离线数据分析、Hadoop。

Spark的四大优点

1. 快:与Hadoop的MapReduce相比，Spark基于内存的运算要快100倍以上；而基于磁盘的运算也要快10倍以上。Spark实现了高效的DAG执行引擎，可以通过基于内存来高效地处理数据流。
2. 容易使用:Spark支持Java、Python和Scala的API，还支持超过80种高级算法，使用户可以快速构建不同应用。而且Spark支持交互式的Python和Scala的Shell，这意味着可以非常方便的在这些Shell中使用Spark集群来验证解决问题的方法，而不是像以前一样，需要打包、上传集群、验证等。这对于原型开发非常重要。
3. 通用性：Spark提供了统一的解决方案。Spark可以用于批处理、交互式查询（通用Spark SQL）、实时流处理（通过Spark Streaming）、机器学习（通过Spark MLlib）和图计算（通过Spark GraphX）。
   这些不同类型的处理都可以在同一应用中无缝使用。Spark统一的解决方案非常具有吸引力，毕竟任何公司都想用统一的平台处理问题，减少开发和维护的人力成本和部署平台的物理成本。当然还有，作为统一的解决方案，Spark并没有以牺牲性能为代价。相反，在性能方面Spark具有巨大优势。

背景

房地产数据是非常有价值的数据资源，可以帮助政府监管房地产市场、协助开发商决策、指导银行贷款和投资决策等。而 Spark 作为大数据处理框架，可以处理大规模的房地产数据，进行数据分析、预测和挖掘，为房地产业提供更多的洞察和启示。

数据介绍

数据是58同城网站爬虫爬取的数据并在python中数据清洗 导入到mysql数据库,数据如下

---

指标介绍

1.城区和街道进行数量统计，分析房产分布和热门地区。

主要步骤介绍

1. 地址信息分离： 通过split函数将原始地址字段分离为不同的部分，例如城市、社区和详细地址，并将这些信息分别存储在新的列中，如“partition”、“Community”和“DetailedAddress”。

2. 统计城区数量： 使用distinct和count函数统计不同城区的数量，得到城区的唯一计数。

3. 计算热门城区和街道： 通过对城区和详细地址进行分组统计，计算每个城区和街道的房产数量。热门城区是按房产数量降序排列的所有城区列表，而热门街道是按房产数量降序排列并选择前五的街道列表。

4. 数据写回： 提供了可选的代码片段，指示如何将分离后的结果写回到数据库，使用了write.format和相应的数据库连接选项。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
 
object AddressData {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("AddressDataSeparation")
      .master("local[*]")
      .getOrCreate()
 
    // 从数据库读取数据
    val jdbcDF = spark.read.format("jdbc")
      .option("url", "jdbc:mysql://xxxxxxxxxx:3306/house?useUnicode=true&characterEncoding=utf8&useSSL=false")
      .option("dbtable", "housedata2")
      .option("user", "root")
      .option("password", "")
      .load()
 
    // 分离地址信息
    val separatedDF = jdbcDF.withColumn("partition", split(col("Address"), "-").getItem(0))
      .withColumn("Community", split(col("Address"), "-").getItem(1))
      .withColumn("DetailedAddress", split(col("Address"), "-").getItem(2))
 
    // 统计城区数量
    val distinctCityCount = separatedDF.select("partition").distinct().count()
 
    // 计算热门城区
    val popularCities = separatedDF.groupBy("partition").count().orderBy(desc("count"))
 
    // 计算热门街道
    val popularStreets = separatedDF.groupBy("DetailedAddress").count().orderBy(desc("count")).limit(5)
 
 
 
    println("城区数量:",distinctCityCount)
 
    //热门城区
    popularCities.show()
 
    //热门街道
    popularStreets.show()
 
     //显示分离结果
     separatedDF.show()
 
     //可以选择将分离后的结果写回数据库
     separatedDF.write.format("jdbc")
       .option("url", "jdbc:mysql://your_mysql_host:port/your_database")
       .option("dbtable", "new_table")
       .option("user", "your_username")
       .option("password", "your_password")
       .save()
 
    spark.stop()
  }
}

2.分析房产数据表中不同建造年份的房产数量情况

主要步骤介绍

1. 提取年份信息： 通过对建造时间字段进行处理，使用 substring 函数提取了年份信息，并创建了一个名为 "Year" 的新列。

2. 按年份分组计数： 对提取出的年份信息进行分组，并使用 groupBy 和 count 方法计算每个年份出现的次数，即统计了每个年份对应的房产数量。

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
 
object ConstructionYear {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("ConstructionYearStatistics")
      .master("local[*]")
      .getOrCreate()
 
    // MySQL连接信息
    val mysqlHost = "xxxxxxxxxxxx"
    val mysqlPort = 3306
    val database = "house"
    val tableName = "housedata2"
    val username = "root"
    val password = ""
 
    // 读取MySQL数据库的建造时间信息
    val jdbcUrl = s"jdbc:mysql://$mysqlHost:$mysqlPort/$database?useUnicode=true&characterEncoding=utf8&useSSL=false"
    val connectionProperties = new java.util.Properties()
    connectionProperties.put("user", username)
    connectionProperties.put("password", password)
    val df = spark.read.jdbc(jdbcUrl, tableName, connectionProperties)
 
    // 提取年份信息并分组计数
    val yearCounts = df.withColumn("Year", substring(col("Constructiontime"), 1, 4))
      .groupBy("Year")
      .count()
 
    // 打印结果
    yearCounts.show()
 
    spark.stop()
  }
}

3.分析不同地区、楼层和户型的房产平均单价和总价的计算，高价房产的识别（价格超过 100 万），以及不同户型房产数量占总量的比例。

主要步骤介绍

1. 按地区、楼层或户型分组计算平均单价和总价：

   • 使用 groupBy 对地区、楼层和户型进行分组。
   • 使用 agg 函数计算每个组的平均单价和总价。

2. 识别高价房产：

   • 过滤出价格高于 100 万的房产数据。

3. 计算不同户型的房产数量占总量的比例：

   • 通过 groupBy 对不同户型进行分组。
   • 使用 count 统计每种户型的数量。
   • 使用 withColumn 计算每种户型数量在总房产数量中的比例。

import org.apache.spark.sql.{SparkSession, functions => F}
import org.apache.spark.sql.functions._
object RealEstateAnalysis {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("Real Estate Analysis")
      .master("local[*]")
      .getOrCreate()
 
    import spark.implicits._
    // 从数据库读取数据
    val jdbcDF = spark.read.format("jdbc")
      .option("url", "jdbc:mysql://xxxxxxxxx:3306/house?useUnicode=true&characterEncoding=utf8&useSSL=false")
      .option("dbtable", "housedata2")
      .option("user", "root")
      .option("password", "")
      .load()
 
    // 分离地址信息
    val separatedDF = jdbcDF.withColumn("City", split(col("Address"), "-").getItem(0))
      .withColumn("Community", split(col("Address"), "-").getItem(1))
      .withColumn("DetailedAddress", split(col("Address"), "-").getItem(2))
 
 
    //  1.按地区、楼层或户型分组计算平均单价和总价
    val averagePriceAnalysis = separatedDF
      .groupBy($"City", $"Floor", $"Housetype")
      .agg(
        F.avg($"Price").as("Average Price"),
        F.sum($"Price").as("Total Price")
      )
 
    // 2. 识别高价房产（这里假设阈值是100万）
    val highValueProperties = jdbcDF.filter($"Price" > 100)
 
    // 3. 计算不同户型的房产数量占总量的比例
    val housetypeComposition = jdbcDF
      .groupBy($"Housetype")
      .count()
      .withColumn("Percentage", $"count" / F.sum("count").over())
 
 
//    // 显示结果
     averagePriceAnalysis.show()
    highValueProperties.show()
    housetypeComposition.show()
 
 
 
 
    // 停止Spark会话
    spark.stop()
  }
}

 

---

总结

这些代码可以展示了如何使用Spark进行房地产数据分析。从分离地址信息到计算平均价格和高价房数量，再到不同户型的房产比例，这些指标为理解房产市场提供了重要见解。通过数据分析，可以更深入了解房产市场的地区趋势、价格分布和市场结构。这些工具和指标为房地产决策者提供了更全面的数据支持，帮助他们做出更明智的决策。