java操作gaussDB数据库

package com.shiwusuo.ReadHdfsToClickHouse.gauss

import java.sql.{Connection, DriverManager}
import java.util.Properties

import org.apache.spark.sql.DataFrame

import scala.collection.mutable.ArrayBuffer;

object GaussDBUtils {

  val urls = application.gaussURL //数据库URL
  val username = application.gaussName //用户名
  val password = application.gaussPass //密码
  val database = application.gaussDatabase
  val gaussCreateDatabase = application.gaussCreateDatabase
  //val driver = "org.postgresql.Driver"
  val driver = application.driver
  var conn: Connection = null;

  def getConnection(): Connection = {
    try {
      Class.forName(driver)
      conn = DriverManager.getConnection(urls, username, password)
      conn
    } catch {
      case e: Throwable => e.printStackTrace()
        println("连接gaussDB失败===请检查原因")
        conn
    }
  }

  //创建表
  def createGaussTable(sql: String): Unit = {
    try {
      val connection = getConnection()
      val statement = connection.createStatement()
      statement.executeUpdate(sql)
      statement.close()
    } catch {
      case e: Exception =>
        e.printStackTrace()
    }

  }

  //判断是否存在该表
  def tableExists(database: String, tablName: String): Boolean = {
    val connection = getConnection()
    val set = connection.getMetaData.getTables(null, database, tablName, null)
    if (set.next()) {
      return true
    } else {
      println("gaussDB中" + tablName + "不存在")
      return false
    }
  }

  //创建数据库
  def createDatabase(): Unit = {
    val connection = getConnection()
    val statement = connection.createStatement()
    statement.executeUpdate(gaussCreateDatabase)
    statement.close()
  }

  //查询 、执行sql语句
  def ReadFromGaussDB(sql: String): Unit = {
    try {
      val connection = getConnection()
      val statement = connection.createStatement()
      val set = statement.executeQuery(sql)
      while (set.next()) {
        val str = set.getString(2)
        println(str)
      }
    }catch {
      case e: Exception =>
        e.printStackTrace()
    }
  }

  //获取gaussDB中所有表名
  def getTablesByGaussDB(db: String) :ArrayBuffer[String]={
    val tablesList = new ArrayBuffer[String]()
    val connection = getConnection()
    val set = connection.getMetaData.getTables(null,db,null,null)
    while (set.next()) {
      val str: String = set.getString("TABLE_NAME")
      tablesList += str
      // println(str)
    }
    tablesList
  }

  //根据resourceId删除gaussDB数据
  def deleteDataByResourceId(sql:String): Unit ={
      val connection = getConnection()
      val statement = connection.createStatement()
      statement.executeUpdate(sql)
  }



  //插入gaussDB库中
  def insertgaussDBTable(table: String, df: DataFrame): Unit = {
    val connectionProperties = new Properties()
    connectionProperties.setProperty("user", username)
    connectionProperties.setProperty("password", password)
    println("+++++++开始写入GaussDB++++++" + table + "共有" + df.count())
    df.write.mode("append").option("batchsize", "50000").option("isolationLevel", "NONE").
      option("numPartitions", "1").jdbc(urls, table, connectionProperties)
    println("=================GaussDB完成写入========" + table + "====================")
  }


  def main(args: Array[String]): Unit = {
    //val bool = tableExists("gaussdb","test_01087")
    //println(bool)
    //val a = "CREATE TABLE gaussdb.CDPCP\n       (\n         code bigint NULL,\n        ident bigint NULL,\n        len bigint NULL,\n        lineNum1 bigint NULL,\n        lineNum2 bigint NULL,\n        lineNum3 bigint NULL,\n        lineNum4 bigint NULL,\n        lineName varchar(10000000) NULL,\n        begTime bigint NULL,\n        endTime bigint NULL,\n        comDur bigint NULL,\n        meanID varchar(10000000) NULL,\n        siteID varchar(10000000) NULL,\n        unitID varchar(10000000) NULL,\n        taskID varchar(10000000) NULL,\n        guid bigint NULL,\n        storTime bigint NULL,\n        mdSecDeg varchar(10000000) NULL,\n        fileSecDeg varchar(10000000) NULL,\n        secDegPro varchar(10000000) NULL,\n        ipVer bigint NULL,\n        srcAddr varchar(10000000) NULL,\n        dstAddr varchar(10000000) NULL,\n        srcPort bigint NULL,\n        dstPort bigint NULL,\n        protNum bigint NULL,\n        srcAddrV6 varchar(10000000) NULL,\n        dstAddrV6 varchar(10000000) NULL,\n        srcLoc varchar(10000000) NULL,\n        dstLoc varchar(10000000) NULL,\n        srcISP varchar(10000000) NULL,\n        dstISP varchar(10000000) NULL,\n        srcAS varchar(10000000) NULL,\n        dstAS varchar(10000000) NULL,\n        protInfo varchar(10000000) NULL,\n        linkInfo varchar(10000000) NULL,\n        protType varchar(10000000) NULL,\n        protName varchar(10000000) NULL,\n        mulRouFlag bigint NULL,\n        intFlag bigint NULL,\n        strDirec bigint NULL,\n        pktNum bigint NULL,\n        payLen bigint NULL,\n        hashId bigint NULL,\n        sessionID varchar(10000000) NULL,\n        resourceId varchar(10000000) NULL\n       )"
    //createGaussTable(a)
    //createtable
   // ReadFromGaussDB("select * from test_0110.user")
    //val strings = getTablesByGaussDB("test_0301")
    //strings.foreach(println)
    //DELETE FROM test_0110.AOE WHERE resourceId = 518824231216091205
    val b ="DELETE FROM test_0110.\"ANS\" WHERE \"resourceId\" = '518824231216091205'"
    println(b)
    deleteDataByResourceId(b)
  }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133