初识Spark

export SPARK_MASTER_IP=SparkIp1  #实际按照机器配置
export SPARK_MASTER_PORT=7077
export SPARK_WORKER_CORES=1
export SPARK_WORKER_INSTANCES=1
export SPARK_WORKER_MEMORY=1g

 SparkIp1
SparkIp2
SparkIp3

sbin/start-master.sh

sbin/start-slave.sh SparkIp1:7077 #这里需要每台slave机器都指定master

/**
    * @param poi_id poi序号
    * @param poi_title 标题
    * @param poi_address 地点
    * @param poi_lon 经度
    * @param poi_lat 维度
    * @param poi_city 城市代码号
    * @param poi_category_name poi类型代号
    * @param poi_checkin_num poi签到次数
    * @param poi_photo_num poi签到照片数量
    */
  case class Poi(poi_id: String, poi_title: String, poi_address: String, poi_lon: Double, poi_lat: Double,
                 poi_city: String, poi_category_name: String, poi_checkin_num: Long, poi_photo_num: Long)

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

    System.setProperty("hadoop.home.dir", "E:\\tools\\hadoop\\hadoop-2.7.2")
    val warehouseLocation = "file:///"

    val spark = SparkSession.builder().master("local").appName("Spark SQL").config("spark.sql.warehouse.dir", warehouseLocation).getOrCreate()

    val provinceVec = Vector("anhui.csv","beijing.csv")

    provinceVec.foreach((province:String) => runInferSchema(spark, province))

  }

  private def runInferSchema(spark: SparkSession, file: String): Unit = {

    println("====================================" + file + "====================================")

    import spark.implicits._

    val poiDF = spark.sparkContext
      .textFile("hdfs://127.0.0.1:9000/weibo_data/" + file)
      .map(_.split(",")).filter(line => line.length==9)
      .map(attr => Poi(attr(0).trim(), attr(1).trim(), attr(2).trim(), attr(3).trim().toDouble, attr(4).trim().toDouble, attr(5).trim(),
          attr(6).trim(), attr(7).trim().toLong, attr(8).trim().toLong))
      .toDF()

    poiDF.createOrReplaceTempView("poi")

    //poi数量，签到数量，照片数量统计
    val sql_sum_result = spark.sql("select count(*) as sum_poi_count, sum(poi_checkin_num) as sum_poi_checkin_num, sum(poi_photo_num) as sum_poi_photo_num from poi")
    //sql_sum_result.show(1)

    //各个地区签到数量统计
    val sql_group_by_checkin_num = spark.sql("select sum(poi_checkin_num) as sum_poi_checkin_num, poi_city from poi group by poi_city")
    //sql_group_by_checkin_num.show()

    //热门签到地点统计
    val sql_group_by_poi_address_num = spark.sql("select count(*) as count, poi_address from poi group by poi_address")
    //sql_group_by_poi_address_num.show(5)

    //签到类型统计
    val sql_group_by_poi_category_name_num = spark.sql("select count(*) as count, poi_category_name from poi group by poi_category_name")
    //sql_group_by_poi_category_name_num.show(5)

    //签到位置统计
    val sql_group_by_location_num = spark.sql("select count(*) as count, poi_lon, poi_lat from poi group by poi_lon, poi_lat")
    sql_group_by_location_num.show(5)

  }

      var sparkConf = new SparkConf().setAppName("SparkStreamReadFile").setMaster("local[2]")
      var sc = new StreamingContext(sparkConf, Seconds(20))

      var lines = sc.textFileStream("D:\\log\\")
      var words = lines.flatMap(_.split(" "))
      var wordConts = words.map(x => (x,1)).reduceByKey(_ + _)

      wordConts.print()
      sc.start()
      sc.awaitTermination()