专栏首页大数据仓库建设Scala JDBC 查询和更新MySQL

Scala JDBC 查询和更新MySQL

package com.bi

import java.sql.{Connection, DriverManager, Timestamp}
import java.util.Calendar


/**
  * Created by xxx on 2017/6/28.
  */
object MySqlConn {

  // for test env
  val mysqlConfTest = collection.mutable.Map(
    "driver" -> "com.mysql.jdbc.Driver",
    "url" -> "jdbc:mysql://192.168.18.106:3306/rpt",
    "username" -> "test",
    "password" -> "test"
  )

  // for prod env
  val mysqlConfProd = collection.mutable.Map(
    "driver" -> "com.mysql.jdbc.Driver",
    "url" -> "jdbc:mysql://hostname:3306/rpt?autoReconnect=true",
    "username" -> "xxxx",
    "password" -> "xxxxx"
  )

  /**
    * 创建mysql连接
    * @return
    */
  def getMysqlConn(): Connection = {
    Class.forName(mysqlConfTest("driver"))
    DriverManager.getConnection(mysqlConfTest("url"), mysqlConfTest("username"), mysqlConfTest("password"))
  }

  /**
    * 查询所有漏斗
    * @return
    */
  def getFunnels(dateStr: String): collection.mutable.Map[Int, (String, Int, Int ,String ,String)] = {
    // connect to the database named "mysql" on the localhost
    val conn = getMysqlConn

    val funnels = collection.mutable.Map[Int, (String, Int, Int ,String ,String)]()

    try {
      // create the statement, and run the select query
      val statement = conn.createStatement()
      val sql = s"""select id,funnel_name,status,done,start_time,end_time
                  |from rpt_funnel_manage
                  |where status = 1
                  |and (done in (0,3) or (done = 2 and '$dateStr' <= end_time and '$dateStr' >= start_time))
                  |order by id""".stripMargin
      println(sql)
      val resultSet = statement.executeQuery(sql)
      while (resultSet.next()) {
        val fid = resultSet.getInt("id")
        val funnelName = resultSet.getString("funnel_name")
        val status = resultSet.getInt("status")
        val done = resultSet.getInt("done")
        val startTime = resultSet.getString("start_time")
        val endTime = resultSet.getString("end_time")

        funnels += (fid -> (funnelName, status, done, startTime, endTime))
      }
    } catch {
      case e:Throwable => {
        println("FunnelGO failed! Coursed by getFunnels error")
        e.printStackTrace()
        System.exit(1)
      }
    } finally {
      conn.close
    }
    funnels
  }

  def getFunnelLevels(funnelId: Int): collection.mutable.Map[Int, List[Int]] = {
    // connect to the database named "mysql" on the localhost
    val conn = getMysqlConn

    val funnel = collection.mutable.Map[Int, List[Int]]()

    try {
      // create the statement, and run the select query
      val statement = conn.createStatement()
      val sql = s"select * from rpt_funnel_manage_level where fid = $funnelId order by level desc"
      val resultSet = statement.executeQuery(sql)
      while (resultSet.next()) {
        val fid = resultSet.getInt("fid")
        val level = resultSet.getInt("level")
        val page_id = resultSet.getInt("page_id")
        val pageIds = List(page_id)

        // 实际调用的是contains(key)
        if(funnel.isDefinedAt(level)){
          val pages = funnel(level)
          // List 多种姿势
          funnel += (level -> pages.++(pageIds))
        } else {
          funnel += (level -> pageIds)
        }
        println("id, funnel_name, page_id = " + fid + ", " + level+ ", " + page_id)
      }
    } catch {
      case e:Throwable => e.printStackTrace()
    } finally {
      conn.close
    }
    funnel
  }

  /**
    * 更新运行状态至rpt_funnel_manage表的done字段
    * 0 未执行 1 执行中 2 执行完成 3 sql执行失败 4 dump执行失败
    * @param funnelId
    * @param runningStatus
    */
  def updateRunningStatus(funnelId: Int, runningStatus: Int): Unit = {
    // create database connection
    val conn = getMysqlConn

    try {
      val ps = conn.prepareStatement("UPDATE rpt_funnel_manage SET done = ? WHERE id = ?")

      // set the preparedstatement parameters
      ps.setInt(1, runningStatus)
      ps.setInt(2, funnelId)

      // call executeUpdate to execute our sql update statement
      val res = ps.executeUpdate()
      ps.close()

      if(res > 0) println(s"UPDATE rpt_funnel_manage SET done = $runningStatus WHERE id = $funnelId success!")
      else println(s"UPDATE rpt_funnel_manage SET done = $runningStatus WHERE id = $funnelId failed!")
    } catch {
      case e:Throwable => e.printStackTrace()
    } finally {
      conn.close
    }
  }

  /**
    * 将sql写入mysql表,同时将sql的运行状态done字段更新为1:运行中
    * @param funnelId
    * @param sqlString
    */
  def updateSql(funnelId: Int, sqlString: String): Unit = {
    // create database connection
    val conn = getMysqlConn

    try {
      val ps = conn.prepareStatement("UPDATE rpt_funnel_manage SET sqlstring = ?,done = 1 WHERE id = ?")

      // set the preparedstatement parameters
      ps.setString(1, sqlString)
      ps.setInt(2, funnelId)

      // call executeUpdate to execute our sql update statement
      val res = ps.executeUpdate()
      ps.close()

      if(res > 0) println(s"UPDATE rpt_funnel_manage SET sqlstring WHERE id = $funnelId success!")
      else println(s"UPDATE rpt_funnel_manage SET sqlstring WHERE id = $funnelId failed!")
    } catch {
      case e:Throwable => e.printStackTrace()
    } finally {
      conn.close
    }
  }

  private def deleteById(conn: Connection, funnelId: Int): Int = {
    var status = 0
    try {
      val sql = s"delete from rpt_funnel_sql where funnelId = $funnelId"
      System.out.println("sql=" + sql)
      val st = conn.createStatement()
      val result = st.executeUpdate(sql)

      //处理结果
      if (result > 0) println(s"delete from rpt_funnel_sql where funnelId = $funnelId success!")
      else println(s"delete from rpt_funnel_sql where funnelId = $funnelId failed!")
      status = result
    } catch {
      case e:Throwable => e.printStackTrace()
    } finally {
      conn.close
    }
    status
  }

  /**
    * 根据 funnelId 删除记录
    * @param funnelId
    */
  def deleteById(funnelId: Int): Int = {
    val connection = getMysqlConn
    val st = connection.createStatement
    var status = 0
    // do database insert
    try {
      val sql = s"delete from rpt_funnel_sql where funnelId = $funnelId"
      System.out.println("sql=" + sql)
      val result = st.executeUpdate(sql)

      //处理结果
      if (result > 0) println("操作成功") else println("操作失败")
      status = result
    } catch {
      case e:Throwable => e.printStackTrace()
    } finally {
      connection.close
    }
    status
  }

  /**
    * 日期
    * @return
    */
  private def getCurrentTimeStamp(): Timestamp = {
    val timeInMillis = Calendar.getInstance.getTimeInMillis
    new Timestamp(timeInMillis)
  }

  def main(args: Array[String]): Unit = {
    val funnelId = args(0).toInt
    MySqlConn.updateRunningStatus(funnelId,0)
  }
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • scala 枚举类

    第2行:通过type定义枚举类型的别名 第3行:我们的枚举从1开始,后续枚举值一次加1

    大数据工程师-公子
  • Scala 字符串插值: s插值

    大数据工程师-公子
  • scala String to Int

    大数据工程师-公子
  • 【算法解析LeetCode by Javascript】23. 合并K个排序链表

    我们只需要把相邻列表进行合并,这样的话我们只需要进行logN次操作就可以把列表归并成一个有序列表

    super.x
  • 原 SparkSQL语法及API

    云飞扬
  • vector方法和使用

    #include<vector>//向量头文件 #include<algorithm>//算法,提供一下函数 using std::cin; using ...

    对弈
  • 微信正在拉低中国社会的整体智商

    用户1756920
  • MySQL索引背后的数据结构及算法原理

    摘要 本文以MySQL数据库为研究对象,讨论与数据库索引相关的一些话题。特别需要说明的是,MySQL支持诸多存储引擎,而各种存储引擎 对索引的支持也各不相同,因...

    wangxl
  • Elasticsearch API 使用介绍

    API文档: https://www.elastic.co/guide/en/elasticsearch/reference/current/docs.htm...

    shaonbean
  • 为什么有些语言比别的快?

    来自Ars Technica的文章评论了影响编程语言速度的各个方面。Ars这个网站虽然自称技术网站,但编程方面的文章一般比较浅,这篇也不例外。虽然文字很长,但无...

    钱曙光

扫码关注云+社区

领取腾讯云代金券