by : digitalevers kafka, scala, 大数据

模拟点击-Akka定时向Kafka发送模拟数据（定时器的另类实现）

Spark的流式计算经常需要模拟流数据，这时就需要使用定时器每隔一秒或每隔几秒向Kafka写入模拟数据，模拟数据的方法及发送策略的制定可以参考这一篇，大体思想就是设置一天模拟的总数据量，然后使用一定的算法分解到每一秒中，再把分解好的数字序列存放到一个大的数组中，最后使用定时器从头到尾扫描这个数组，逐一取出再组装相应的模拟数据量就是了。

可以使用原生 Java 的 Timer 定时调度工具来实现这个定时器，但是这里采用一种另类的方式来实现——采用Akka的actor来实现。整体来说比较简单，直接上代码

package com.datacrafts.digitalevers.simulationClick

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.datacrafts.digitalevers.simulationClick.common._
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration
import scala.concurrent.duration.FiniteDuration

class sendClick extends Actor{

  override def receive: Receive = {
    case "start" =>{
      println("客户端启动")
      import context.dispatcher
      context.system.scheduler.scheduleWithFixedDelay(FiniteDuration(0,duration.MILLISECONDS),FiniteDuration(1000,duration.MILLISECONDS), self, sendData2Kafka)
    }
    case sendData2Kafka=>{
      println("客户端向Kafka发送数据~")
    }
  }

}


object sendClick {
  def main(args: Array[String]): Unit = {
    val host = "127.0.0.1"
    val port = 10006

    val config = ConfigFactory.parseString(s"""
                                              |akka.actor.provider="akka.remote.RemoteActorRefProvider"
                                              |akka.actor.allow-java-serialization = on
                                              |akka.remote.artery.canonical.hostname=$host
                                              |akka.remote.artery.canonical.port=$port""".stripMargin)

    val actorSystem = ActorSystem("work",config)
    val sparkWorkRef: ActorRef = actorSystem.actorOf(Props(new sendClick),"sparkWork")

    sparkWorkRef ! "start"
  }
}

这里采用的是 Akka2.6.8，其他版本的配置参数可能会有些许不同，读者可自行查阅文档。host 和 port 可配可不配，这里还是配置上了，如果不配置的话，Akka会默认当前主机，同时随机打开一个端口供 actor 使用。

如代码所见，这个例程去掉了 Akka 与服务端的交互，而 Akka 的一般应用场景就是与服务端的交互。这里只利用到 actor 对自身发送消息这一机制

当actor启动后，便启动定时器每隔一秒向自身发送一个 sendData2Kafka 对象消息，这个空消息对象定义如下

package com.datacrafts.digitalevers.simulationClick.common

case class sendData2Kafka(){}

当 actor 接受到自身的这个对象消息后，便会执行 case sendData2Kafka 内的代码，这里便可以实现向 Kafka 写数据的逻辑。

以上便是整个流程，如此简单。

Leave a comment
by : digitalevers spark, 大数据, 算法

Spark实时流计算去重探索

Spark离线处理去重问题不难，把所有历史数据读进内存，然后使用Spark的各种操作算子进行统计去重即可。但是实时流计算就没那么方便了，经过一段时间的研究，摸索出了两种可供参考的方法，如下。

数据量小的情况下，读取所有的数据唯一标识符

比如已有100万手机设备的数据，现在有一个设备信息进来，如何判断这个设备数据是新用户还是旧用户。我们选取IMEI这个唯一标识来辨别设备，先把这100万的设备IMEI全部用Spark读取进来，然后实时与这个新的设备信息进行匹配，便可得知是新设备还是旧设备（在没有大数据处理框架之前，这个匹配工作是由数据库（Mysql）执行查询SQL语句来完成）。

这个思路与离线计算统计去重基本一致，但是Spark需要维护这个庞大的历史设备信息库，如果有一个新增设备，要立即加到历史设备信息库上去。随着时间的推移，这个维护量也会越来越大。

数据量很庞大的情况下

如果历史设备信息已经很庞大，比如已经超过1000万，这时再继续维护一个IMEI库就会有巨大的内存开销，既然每个IMEI都不一样，那么我们是不是可以转换一种思路，通过一种hash算法将每个IMEI映射成一个整数，然后再构造一个巨大的 bitmap 表，通过 bitmap 表的一个位(bit)来对应这个整数，熟悉 bitmap 算法的人都知道，这种去重方法是非常节约内存的。一个 bit 位即代表一个设备，通过查找这个位的值为1还是0来判别是否新设备。（1是旧设备，0是新设备）

如果使用 Scala 构造一个10万整数的数组，可以表达 10万 * 32 = 320 万设备信息（一个整数类型拥有32位）。而这10万数组占用320万/8 = 40万个字节，40万/1024/1024 = 0.38 M，不到0.4M的内存占用，可以说是非常高效了。即使数据量以亿为量级，内存占用也不过几十M而已

再回到之前提到的hash算法，也的确是存在这种字符串到整数的哈希算法的，而且还不止一种，经过多方测评，其中的BKDRHash算法最为优秀，详情可以参考 https://blog.csdn.net/hzhsan/article/details/25552153?utm_source=blogxgwz1

下面使用Scala对 bitmap 算法进行具体实现

沿袭数组索引从0开始的惯例，我们也使用第1位留空，第2位才表示设备1，第3位表示设备2，依次类推。设备编号即为这个bit位的索引。举个栗子，如果一个设备的IMEI（不一定必须IMEI，也可以是其他参数），通过BKDRHash算法映射到了1001这个整数，那么这个数组从左往右数第1002位便表示这个设备。

因为像 Scala 这样的高级语言，并没有直接提供数组内单个位的索引功能，只提供通过 arr(0) arr(1) 这样对数组元素的访问机制，也就是说只提供以32位为单位的一个索引功能。接下来要做的第一件事就是确定目标位在数组中的索引index是多少

从图中可以看出，array(0) 可以表示整数 0 – 31

array(1) 表示 32 – 63 array(2) 表示 64 – 95

可以看出以32为一个单位，目标整数除以32取整即可得到这个index值

(1001/32).toInt 为 31，所以这个整数落在array(31)内

接下来计算1001在array(31)内的偏移量。这个偏移量其实就是对32取模之后的值，1001 % 32 = 9 。然后进行位移 1 << 9，找到了这个位置，就可以对其进行位运算。所以最后的置1位运算公式看起来是

arr(n/32) |= 1 << (n%32)

转换成位运算符等同于

arr(n >> 5) |= 1 << (n & 0x1F)

上面是设置这个位的操作，还有查询这个位的操作

(arr(n>>5) & (1<<(n&0X1F))) != 0

如果为true，表示该位已被设置，应用层则表示该设备是旧设备，否则为false,为新增设备

重置位操作

arr(n>>5) &= (~(1<<(n&0X1F)))

以上便是实时去重的全部思路，如果读者有更好的实现方案，欢迎探讨。

Leave a comment
by : digitalevers scala, 大数据, 教程&知识

庖丁解牛-Scala函数柯里化实现读取文件

package com.datacrafts.digitalevers.curry

import java.io.{BufferedReader, Closeable, File, FileReader}
import com.esotericsoftware.minlog.Log

/**
 * 这段代码使用函数柯里化的功能实现了读取文件源并将其转为List数据
 * 函数柯里化的精髓是使得函数实现在依赖其他函数的情况下实现可插拔，从而在函数与函数之间的调用上进行解耦
 * 比如这段代码可以在getLines底层函数上只保留 filename 一个参数，在函数体内直接调用 isReadable 和  closeStream方法来进行
 * 周边逻辑的实现，一样可以达到效果，但是如果将来需要修改 isReadable 和  closeStream 的功能，不得不修改这两个方法的源代码
 * 如果采用柯里化的形式 可以重新编写一个 isReadable 和 closeStream 的实现，在上层调用的时候直接将新函数作为参数传入即可
 * 从而符合ocp原则。
 */
object FileReader {

  /**
   * 柯里化后的函数
   * @param filename
   * @param isFileReadable
   * @param closableStream
   * @return
   */
  def getLines (filename: String) (isFileReadable: (File) => Boolean) (closableStream: (Closeable) => Unit): String = {
    val file = new File(filename)
    if (isFileReadable (file) ) {
      val readerStream = new FileReader(file)
      val buffer = new BufferedReader(readerStream)
      try {
        var content = ""
        var str = ""
        var isReadOver = false
        while (! isReadOver) {
          str = buffer.readLine ()
          if (str == null) {
            isReadOver = true
          } else {
            content += str
          }
        }
        content
      } finally {
        closableStream (buffer)
        closableStream (readerStream)
      }
    } else {
      ""
    }
  }

  /**
   * 判断文件是否可读
   * @param file
   * @return
   */
  def isReadable (file: File) = {
    if (null != file && file.exists() && file.canRead() ) {
      true
    } else {
      false
    }
  }

  /**
   *关闭打开的文件流
   * @param stream
   */
  def closeStream (stream: Closeable) {
    if (null != stream) {
      try {
        stream.close
      } catch {
        case ex:Throwable => Log.error( "[" +this.getClass.getName + ".closeStream] ", ex.getMessage)
      }
    }
  }
}
Leave a comment
by : digitalevers scala, 大数据, 教程&知识

伸缩自如-Scala实现稀疏矩阵解压缩

package com.datacrafts.digitalevers.sparseMatrix

import java.io.FileWriter

import com.datacrafts.digitalevers.curry.FileReader.{closeStream, getLines, isReadable}
import com.google.gson.Gson
import scala.collection.mutable.ArrayBuffer


/**
 * 此例程提供Scala对稀疏矩阵进行压缩解压模拟的操作方法
 * 通用方式
 */
object normal{

  implicit val filePath = ""

  def main(args: Array[String]): Unit = {
    /*var test = Array.ofDim[Int](11,3)
        test(5)(2) = 11
        test(3)(1) = 5
        var resJson = zipMatrix(test)(checkMatrix)("json.txt")*/
    val arr = unzipMatrix("json.txt")

  }

  /**
   * 压缩稀疏矩阵并将其保存到文件中
   */
  def zipMatrix(inArray:Array[Array[Int]])(check:(Array[Array[Int]])=>Array[Int])(implicit filePath:String) = {
    val newMatrix = ArrayBuffer[Array[Int]]()
    val res:Array[Int] = check(inArray)
    newMatrix.append(Array(res(0),res(1),0))
    for(i <- 0 until inArray.length){
      for(j <- 0 until inArray(i).length){
        if(inArray(i)(j) != 0){
          newMatrix.append(Array(i,j,inArray(i)(j)))
        }
      }
    }

    //print(filePath)
    val arr = newMatrix.toArray   //ArrayBuffer -> Array -> json
    val gson = new Gson()
    val json = gson.toJson(arr)
    if(filePath.isEmpty == true){
      json
    } else {
      save2File(filePath,json)
    }
  }

  /**
   * 从文件中读取数据并解压成稀疏矩阵
   */
  def unzipMatrix(filePath:String) = {
    val content = getLines(filePath) (isReadable) (closeStream)
    if(content == ""){
      throw new Exception("文件内容为空")
    }
    val res = ArrayBuffer[Array[Int]]()
    val gson = new Gson()
    val arr = gson.fromJson(content,classOf[Array[Array[Int]]])
    for(i <- 0 until arr.length){
      if(i == 0){
        for(j <- 0 until arr(i)(0)){
          res.append(Array.ofDim[Int](arr(i)(1)))
        }
      } else {
        res(arr(i)(0))(arr(i)(1)) = arr(i)(2)
      }
    }
    res.toArray
  }

  /**
   * 检测数组的容积（长和宽）
   * @param inArray
   * @return
   */
  def checkMatrix(inArray:Array[Array[Int]]):Array[Int] = {
    if(inArray.length > 0){
      val i = inArray.length
      val j = inArray(0).length
      Array(i,j)
    } else {
      Array(0,0)
    }
  }

  /**
   * 保存字符串到文件中
   * @param filePath 文件路径
   * @param content   需要保存的字符内容
   * @return 保证成功 true 保存失败 false
   */
  def save2File(filePath:String,content:String): Boolean = {
      try {
        val out = new FileWriter(filePath, true)
        out.write(content)
        out.close()
        true
      } catch {
        case ex: Throwable => {
          false
        }
      }
  }
  ////////////////
}

下面是泛型方式，在开发泛型方式的时候有一个小插曲

最开始使用Play Framework框架中的json解析库（play.api.libs.json）无法解析泛型的数组结构，更换到gson解析库，便可正常解析

package com.datacrafts.digitalevers.sparseMatrix

import com.datacrafts.digitalevers.curry.FileReader.{closeStream, getLines, isReadable}
import com.datacrafts.digitalevers.sparseMatrix.normal.{save2File, unzipMatrix}
import com.google.gson.Gson
import scala.collection.mutable.ArrayBuffer

/**
 * 此例程提供Scala对稀疏矩阵进行压缩解压模拟的操作方法
 * 泛型方式
 */
object generic{

  implicit val filePath = ""

  def main(args: Array[String]): Unit = {
        var test = Array.ofDim[Double](11,3)
        test(5)(2) = 11.1
        test(3)(1) = 5.2
        var res = zipMatrix(test)(checkMatrix)("json_generic.txt")
    val arr = unzipMatrix("json_generic.txt")
    for(line <- arr){
      for(elem <- line){
        print(elem.toString + " ")
      }
      println()
    }
  }

  /**
   * 压缩稀疏矩阵并将其保存到文件中
   */
  def zipMatrix[T](inArray:Array[Array[T]])(check:(Array[Array[T]])=>Array[Int])(implicit filePath:String) = {
    val newMatrix = ArrayBuffer[Array[Any]]()
    val res:Array[Int] = check(inArray)
    newMatrix.append(Array(res(0),res(1),0))
    for(i <- 0 until inArray.length){
      for(j <- 0 until inArray(i).length){
        if(inArray(i)(j) != 0){
          newMatrix.append(Array(i,j,inArray(i)(j)))
        }
      }
    }
    val arr = newMatrix.toArray
    val gson = new Gson()
    val json = gson.toJson(arr)
    if(filePath.isEmpty == true){
      json
    } else {
      save2File(filePath,json)
    }
  }

  /**
   * 因为泛型的压缩数组数据类型不确定
   * 所以这个隐式函数提供给 unzipMatrix 作隐式类型转换
   * @param in
   * @return
   */
  implicit def any2Int(in:Any):Int = {
    in.toString.toDouble.toInt
  }

  /**
   * 从文件中读取数据并解压成稀疏矩阵
   */
  def unzipMatrix(filePath:String) = {
    val content = getLines(filePath) (isReadable) (closeStream)
    if(content == ""){
      throw new Exception("文件内容为空")
    }
    val res = ArrayBuffer[Array[Any]]()
    val gson = new Gson()
    val arr = gson.fromJson(content,classOf[Array[Array[Any]]])
    for(i <- 0 until arr.length){
      if(i == 0){
        for(j <- 0 until arr(0)(0)){
          val line = ArrayBuffer[Any]()
          for(k <- 0 until arr(0)(1)){
            line.append(0)
          }
          res.append(line.toArray)
        }
        ///初始化
      } else {
        res(arr(i)(0))(arr(i)(1)) = arr(i)(2)
      }
    }
    res.toArray
  }

  /**
   * 检测数组的容积（长和宽）
   * @param inArray
   * @tparam T
   * @return
   */
  def checkMatrix[T](inArray:Array[Array[T]]):Array[Int] = {
    if(inArray.length > 0){
      val i = inArray.length
      val j = inArray(0).length
      Array(i,j)
    } else {
      Array(0,0)
    }
  }

}
Leave a comment
by : digitalevers scala, 大数据, 教程&知识

庖丁解牛-记Scala函数柯里化读取文件并转化为List

package com.datacrafts.digitalevers.curry

import java.io.{BufferedReader, Closeable, File, FileReader}
import com.esotericsoftware.minlog.Log

/**
 * 这段代码使用函数柯里化的功能实现了读取文件源并将其转为List数据
 *
 * 函数柯里化的精髓是使得函数在依赖其他函数的情况下实现可插拔，从而在函数与函数之间的调用上进行解耦
 * 比如这段代码也可以在getLines底层函数上只保留 filename 一个参数，在函数体内直接调用 isReadable 和  closeStream 方法来进行
 * 周边逻辑的实现，可以达到同样的效果，但是如果将来需要修改 isReadable 和  closeStream 的功能，不得不修改这两个方法的源代码
 * 如果采用柯里化的形式 可以重新编写一个 isReadable 和 closeStream 的实现，在上层调用的时候直接将新函数作为参数传入即可
 * 从而符合ocp原则。
 */
object File2List {
  def main(args: Array[String]): Unit = {
      var list = readFile("1.txt")
      print(list)
  }

  def readFile (filename: String): List[String] = {
    getLines(filename) (isReadable) (closeStream)
  }

  /**
   * 柯里化后的函数
   * @param filename
   * @param isFileReadable
   * @param closableStream
   * @return
   */
  private def getLines (filename: String) (isFileReadable: (File) => Boolean) (closableStream: (Closeable) => Unit): List[String] = {
    val file = new File(filename)
    if (isFileReadable (file) ) {
      val readerStream = new FileReader(file)
      val buffer = new BufferedReader(readerStream)
      try {
        var list: List[String] = List ()
        var str = ""
        var isReadOver = false
        while (! isReadOver) {
          str = buffer.readLine ()
          if (str == null) isReadOver = true
          else list = str :: list
        }
        list.reverse
      } finally {
        closableStream (buffer)
        closableStream (readerStream)
      }
    } else {
      List ()
    }
  }

  /**
   * 判断文件是否可读
   * @param file
   * @return
   */
  def isReadable (file: File) = {
    if (null != file && file.exists() && file.canRead() ) {
      true
    } else {
      false
    }
  }

  /**
   *关闭打开的文件流
   * @param stream
   */
  def closeStream (stream: Closeable) {
    if (null != stream) {
      try {
        stream.close
      } catch {
        case ex => Log.error( "[" +this.getClass.getName + ".closeStream] ", ex.getMessage)
      }
    }
  }
}
Leave a comment
by : digitalevers scala, 大数据, 教程&知识

化繁为简—记Scala的继承规则

无论实例对象的类型被定义成基类类型还是子类类型，都优先执行实例化目标类中的方法，若该类中无此方法，则往上寻找去执行基类的该方法

class Base{
  def show():Unit = {
    printf("Base")
  }
}

class Sub extends Base{
  override def show(): Unit = {
    printf("Sub")
  }
}

object Test {
  def main(args: Array[String]): Unit = {
    val sub:Sub = new Sub
    sub.show()
  }
}

上述代码基类和子类分别有一个show方法，子类对其进行了重写，最后的输出结果是打印了 Sub 字符串，可见Sub子类中的show方法得到了执行。关键的来了，即使 val sub:Sub = new Sub 这行改为了

var sub:Base = new Sub

最后打印结果依旧没变，可见印证了上述结论

然后对该段代码稍作修改，注释掉子类的 show 方法，再看一下结果

class Base{
  def show():Unit = {
    printf("Base")
  }
}

class Sub extends Base{
  /*override def show(): Unit = {
    printf("Sub")
  }*/
}

object Test {
  def main(args: Array[String]): Unit = {
    val sub:Sub = new Sub
    sub.show()
  }
}

这是发现打印结果变成了 Sub 字符串，这是因为子类中的确是没有该方法了，不得已去基类查找，最后执行基类的show方法

若基类子类的属性不存在覆盖行为，则方法对属性的获取是就近原则，若存在覆盖行为，则方法永远获取子类的属性

不存在覆盖的行为指属性都由 private 修饰，就近原则指基类的方法获取基类的属性，子类的方法获取子类的属性
class Base{
  private val name = "Base"
  def show():Unit = {
    printf(this.name)
  }
}

class Sub extends Base{
  private val name = "Sub"
}

object Test {
  def main(args: Array[String]): Unit = {
    val sub:Base = new Sub
    sub.show()
  }
}

上述代码执行基类的show方法，由于name属性都被private修饰，不存在覆盖，采取就近原则，最后打印结果是基类的name值——”Base”

class Base{
  val name = "Base"
  def show():Unit = {
    printf(this.name)
  }
}

class Sub extends Base{
  override val name = "Sub"
}

object Test {
  def main(args: Array[String]): Unit = {
    val sub:Base = new Sub
    sub.show()
  }
}

然后对其进行改写，使得name属性被覆写，这时发现虽然执行的还是基类的show方法，但是打印的是子类的name值——”Sub”，印证了这一结论

Leave a comment
by : digitalevers scala, 大数据, 教程&知识

Scala的继承类构造分析

技术环境：Scala2.12.10 因为各版本间可能存在差异，所以其他版本的表现只能由读者自己去验证了，此只作为该Scala版本的验证依据

Scala的继承类构造函数的执行顺序永远遵循下面两个原则

无论入口在哪，首先执行主构造器，再执行辅助构造器

若存在继承关系，则首先执行基类的构造器，再执行子类的构造器

下面的代码可以验证这两个原则

package com.datacrafts.digitalevers.test

object Test {
  def main(args: Array[String]): Unit = {
    val test = new Sub(123,"zhangsan")
  }
}

class Base(){
  println("Base主构造器执行完毕~")
  def this(inAge:Int){
    this
    println("Base第一个辅助构造器执行完毕~")
  }
}

class Sub(inAge:Int) extends Base(inAge:Int){
  var name:String = ""
  var age:Int = 10
  println("Sub主构造器执行完毕~")
  def this(name:String) = {
    this(10)
    this.name = name
    println("Sub第一个辅助构造器执行完毕~")
  }

  def this(age:Int,name:String) = {
    this(age)
    this.age = age
    this.name = name
    println("Sub第二个辅助构造器执行完毕~")
  }
}

最后的输出结果

Base主构造器执行完毕~
Base第一个辅助构造器执行完毕~
Sub主构造器执行完毕~
Sub第二个辅助构造器执行完毕~

这里补充两个注意事项

Scala类的主构造器和辅助构造器之间及辅助构造器与辅助构造器之间都不能存在雷同的参数样式。这里的雷同是指参数个数相同并且参数的类型也一致，因为只要这样的雷同函数发生，Scala的类在实例化的时候就会很迷惑，不知道该何去何从，该执行哪个构造器

如果类之间存在继承关系，既不能在子类的辅助构造器内也不能子类的主构造器内显式调用super来执行基类的构造器，只能由子类的主构造器隐式调用。而基类构造器需要的参数也需要在子类的主构造器上传递过来，如下形式

class Sub(inAge:Int) extends Base(inAge:Int){
………..
}

Leave a comment
by : digitalevers kafka, 大数据, 教程&知识

kafka-manager使用细则

技术环境：CentOS6.5 + kafka-manager2.0 + kafka_2.12-2.5.0

技术背景：为了方便对kafka内部数据进行监控和管理，便找到了kafka-manager，这次使用的是2.0版本，其最新github版本已升至3.0.0.5

下载和安装

kafka-manager由yahoo开源，可以到yahoo的github主页下载已经编译好的二进制以及未编译的源代码
下面是主页地址
https://github.com/yahoo/CMAK/releases

直接可以下载运行的二进制文件

https://github.com/yahoo/CMAK/releases/download/3.0.0.5/cmak-3.0.0.5.zip

需要编译的源代码

https://github.com/yahoo/CMAK/archive/3.0.0.5.zip

如果使用源码，那会是一个漫长的编译过程，整个流程另起篇幅讲解，这里不再讲述，这里只讲解二进制版本

二进制版本可以直接运行，首先需要配置 bin/application.conf 文件，这个配置比较简单，只需要配置一行获取到zookeeper的位置就行了，如下所示

kafka-manager.zkhosts=”localhost:2181″
#kafka-manager.zkhosts=${?ZK_HOSTS}

找到kafka-manager.zkhosts=${?ZK_HOSTS}这一行，注释掉，然后添加一行kafka-manager.zkhosts=”localhost:2181″
因为只有一个kafka节点，使用的是其自带的zookeeper，所以是localhost:2181,如果有一个集群，可以在后面追加。比如
kafka-manager.zkhosts=”ip1:2181,ip2:2181″

也可以通过直接使用ZK_HOSTS环境变量来进行配置，比如
讲ZK_HOSTS=”my.zookeeper.host.com:2181″添加至.bashrc文件中
两种风格看个人喜好，皆可行

配置好后直接启动（保证zookeeper和kafka的服务正常）

bin/kafka-manager -Dconfig.file=conf/application.conf

kafka-manager的默认访问端口是9000，如果不是9000，也可以在启动时通过-Dhttp.port参数来指定具体的端口号，如下所示，便将其修改为8080

bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=8080 &

正常启动后，通过浏览器访问 http://localhost:8080/ 即可看到kafka-manger的界面

参考：https://www.cnblogs.com/frankdeng/p/9584870.html

Leave a comment
by : digitalevers 大数据

Windows下IDEA调试spark程序问题详解

Failed to locate the winutils binary in the hadoop binary path 问题

Windows环境下会出现这样的提示。Linux则不会——我只知道有这样的现象，目前还不知道为何会这样

使用的sbt构建的项目，sbt配置如下

name := "sbt"

version := "0.1"

scalaVersion := "2.12.9"

libraryDependencies ++= Seq(
  "com.fasterxml.jackson.core" % "jackson-annotations" % "2.11.0",
  "org.apache.spark" %% "spark-core" % "3.0.0",
  "org.apache.spark" %% "spark-sql" % "3.0.0",
  "org.apache.spark" %% "spark-streaming" % "3.0.0",
  "org.apache.spark" %% "spark-streaming-kafka-0-10-assembly" % "2.4.3",
  "mysql" % "mysql-connector-java" % "8.0.20",
  "org.apache.kafka" % "kafka-clients" % "2.5.0",
  "com.alibaba" % "fastjson" % "1.2.71"
)

spark样例代码

package _2020_07_25

import org.apache.spark.{SparkConf, SparkContext}

object sparkDemo {
  def main(args: Array[String]): Unit = {
    val sparkconf = new SparkConf().setMaster("local").setAppName("wordCount")
    val sc = new SparkContext(sparkconf)

    printf("init ok ")
  }
}

运行后报出
Failed to locate the winutils binary in the hadoop binary path
的警告提示，解决的办法就是配置一下Windows的HADOOP_HOME环境变量，步骤如下

在 https://github.com/srccodes/hadoop-common-2.2.0-bin 下载该项目的zip包，将其解压到某个指定的目录，比如
C:\Users\Administrator\IdeaProjects\hadoop-common-2.2.0-bin-master (C:\Users\Administrator\IdeaProjects是我的IDEA的工程目录)

新建一个系统环境变量为HADOOP_HOME，将其设置为 C:\Users\Administrator\IdeaProjects\hadoop-common-2.2.0-bin-master 注意不要包含bin目录（它会自动加上bin目录，加上反而会出错）

重启IDEA，该警告提示消失。

参考：https://blog.csdn.net/lepton126/article/details/85987197
https://blog.csdn.net/MASILEJFOAISEGJIAE/article/details/89281982

Leave a comment