import bdqn.sparkdemp.util.ChenIpUtil
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object UDFDemo {
  def main(args: Array[String]): Unit = {
    // 自定义一个函数，实现查询dep表时，deptno返回的是具体的部门信息不是编号信息
    // 使用UDF来实现
    val session: SparkSession = SparkSession.builder()
      .master("local")
      .appName(this.getClass.getSimpleName)
      .getOrCreate()

    val ipDS: Dataset[String] = session.read.textFile("D:\\testLogs\\ip.txt")
    val logDS: Dataset[String] = session.read.textFile("D:\\testLogs\\ipaccess.log")
    import session.implicits._
    val arrRules: Array[(Long, Long, String)] = ipDS.map(t => {
      val str: Array[String] = t.split("\\|")
      (str(2).toLong, str(3).toLong, str(6))
    }).collect()

    val longIpDF: DataFrame = logDS.map(str => {
      // 把ip转换成一个long
      ChenIpUtil.ip2long(str.split("\\|")(1))
    }).toDF("longIp")
    longIpDF.createTempView("t_longIp")

    // 自定义一个UDF函数，实现，接受一个long值，返回一个省份
    // 使用UDF，需要先注册
    session.udf.register("ip2Pro", (ip: Long) => {
      Binarysearch(arrRules, ip)
    })

    // 使用自定义函数
    session.sql("select longIp,ip2Pro(longIp) from t_longIp").show()


  }

  // 二分查找
  def Binarysearch(arr: Array[(Long, Long, String)], ip: Long): String = {
    var left = 0
    var right = arr.size - 1

    while (left <= right) {
      var mid = left + ((right - left) >> 1) // 防止溢出， 并且移位操作更高效
      if (arr(mid)._1 > ip) {
        right = mid - 1
      } else if (arr(mid)._2 < ip) {
        left = mid + 1
      } else {
        return arr(mid)._3 //注意 添加return， scala中只有最后一行才是作为返回值
      }
    }
    null
  }
}