SparkStreaming累加计算单词出现的频率

SparkStreaming累加计算单词出现的频率 SparkStreaming累加计算单词出现的频率

一.需求分析
二，实验环境
三，思路分析
四，编程思想
五.遇见的问题

一.需求分析

服务器中不断产生的数据，通过sparkstreaming的客户端统计服务器端不断产生的相同数据，即就是累计服务器中相同数据出现次数。

二，实验环境

centos7+nc+Spark-2.4.8+windows+idea

三，思路分析

思路：以windows 为客户端，在idea中编写对服务端要统计累加求和的方法,以centos7的Linux为客户端，内存作为数据的产生端，启动nc -l -p 666(链接的端口号）并向Linux中输入自己的数据，
原因：每次客户端程处理服务器端数据银行，会将其结果缓存到检查点中，下一次客户端读入数据并处理数据时回去检查点进行key 值进行查询更新，并且将结果重新更新到检查点，检查点：本质上就是对应的HDFS上的一个目录，将数据写入到该目录以文件的形式将结果保存下来，所以需要先在HDFS上创建检查点对应的目录hdfs dfs -mkdir /spark

四，编程思想

实验步骤：
1.启动idea，编写客户端处理程序：

package com.pmany.Test

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Sparksreaming").setMaster("local[2]")
    val sc = new StreamingContext(conf, Seconds(3))
    //检查点的位置
    sc.checkpoint("hdfs://niit01:9000/spark/checkpoint")
    val v1= sc.socketTextStream("niit01", 7777, StorageLevel.MEMORY_AND_DISK_SER)
    val v2 = v1.flatMap(_.split(" "))

    //累计求和
    val addfun=(currValues:Seq[Int],prevValueState:Option[Int])=>{
    val sum = currValues.sum
    val i = prevValueState.getOrElse(0)
    Some(sum+i)
  }
    val pairs = v2.map(word => (word, 1))
    val v3 = pairs.updateStateByKey(addfun)
    v3.print()
    sc.start()
    sc.awaitTermination()
  }

}

2.运行Iidea客户端的程序，
3.启动Linux，启动nc nc -l -p 6666 '如果没有安装nc :yum install -y nc并且向里面输入数据，一条数据就要执行一次回车：

4,客户端控制台的运行结果：

5.查看检查点数据：

五.遇见的问题

1.链接出错：没有启动 hdfs：start-all.sh或者start-yarn.sh

2.hdfs 处于安全模式;在Hadoop目录关闭安全模式就行bin/hadoop dfsadmin -safemode leave