[大数据]Spark与Scala连接Postgre库

在开发spark过程中，遇到需要连接postgres库的场景，可以通过原生的jdbc去连接，也可以使用spark直连。

1.scala 原生jdbc连接

下面是 通过DriverManager连接pg

try{
// 将用户名密码加入properties
val conn_prop = new Properties()
conn_prop.put("user","xxxx")
conn_prop.put("password","xxxx")
//pg库的连接串：'xxxx'是库名
val conn_url = "jdbc:postgresql://127.0.0.1:5432/xxxx"
//连接
val con:Connection = DriverManager.getConnection(conn_url,conn_prop )
//创建statement
//查询
val query_stm:Statement = con.createStatement()
val rs1= query_stm.excuteQuery("select id,name,money from test_zero where id<1")
while(rs1.next()){
	val r_id = rs1.getInt("id")
	val r_name = rs1.getString("name")
	val r_sex = rs1.getDouble("money")
	println(r_id+","+r_name+","+r_money)
}
//更新
val update_stm:Statement = con.createStatement()
val rs2 = update_stm.excuteUpdate("update test_zero set name='zero_update',money=100000.00 where name='zero'")
}
finally{
   conn.close()
}

下面是 通过Spark直接连接pg,有两种方式

//创建spark
val conf  = new SparkConf().setAppName("conncetPG").setMaster("local[*]")
val spark = SparkSession.builder().config(conf).getOrCreate()
/**
 * 第一种：spark.read.jdbc()
 */
val conn_prop = new Properties()
conn_prop.put("user","xxxx")
conn_prop.put("password","xxxx")
conn_prop.put("driver","org.postgresql.Driver")
val df1 = spark.read.jdbc(url="jdbc:postgresql://127.0.0.1:5432/xxxx",table="test_zero",conn_prop) 
df.select("id","name","money").show()
/**
 * 第二种：spark.read.format("jdbc").option()
 */
 val df2 = spark.read
 		.format("jdbc")
 		.option("url","jdbc:postgresql://127.0.0.1:5432/xxxx")
 		.option("dbtable","test_zero")
 		.option("user","xxxx")
 		.option("password","xxxx")
 		.load()
 //对插叙结果进行过滤操作
 val n_df2 = df2.select("id","name","money").filter("name='zero'")
 //创建临时表，这样可以直接使用spark.sql()
 n_df2.createOrReplaceTempView("tmp_test_zero")
 //执行sparksql
 spark.sql("select * from tmp_test_zero where name like '%zero%'").show()