大数据学习day38----数据仓库01-----区域字典的生成

create table area_dict
as
SELECT
a.BD09_LNG as lng,
a.BD09_LAT as lat,
a.AREANAME as district,
b.AREANAME as city,
c.AREANAME as province

from t_md_areas a 
join t_md_areas b  on a.`LEVEL`=3 and a.PARENTID=b.ID
join t_md_areas c  on b.PARENTID = c.ID

package com._51doit

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.{DataFrame, SparkSession}

object AreaDictGenerator {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession实例
    val spark: SparkSession = SparkSession.builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      .getOrCreate()
    // 创建连接数据库需要的参数
    val probs: Properties = new Properties()
    probs.setProperty("driver", "com.mysql.jdbc.Driver")
    probs.setProperty("user","root")
    probs.setProperty("password", "feng")
    // 以读取mysql数据库的形式创建DataFrame
    val df: DataFrame = spark.read.jdbc("jdbc:mysql://localhost:3306/db_demo1?characterEncoding=UTF-8", "area_dict", probs)
    // 运算逻辑
    import spark.implicits._
    val res: DataFrame = df.rdd.map(row => {
      val lng = row.getAs[Double]("lng")
      val lat = row.getAs[Double]("lat")
      val district = row.getAs[String]("district")
      val city = row.getAs[String]("city")
      val province = row.getAs[String]("province")
      val geoCode: String = GeoHash.withCharacterPrecision(lat, lng, 5).toBase32
      (geoCode, province, city, district)
    }).toDF("geo", "province", "city", "district")
    res.write.parquet("E:/javafile/spark/out11")
  }
}