Spark jdbc postgresql数据库连接和写入操作源码解读
2017-02-10 09:08
363 查看
概述:Spark postgresql jdbc 数据库连接和写入操作源码解读,详细记录了SparkSQL对数据库的操作,通过java程序,在本地开发和运行。整体为,Spark建立数据库连接,读取数据,将DataFrame数据写入另一个数据库表中。附带完整项目源码(完整项目源码github)。
1.1. 在postgreSQL中的postgres用户下,创建
1.2. 在
查看数据库写入结果。
其中
库驱动路径,命令执行要在项目的根目录中(/home/xiaolei/Data/GS/Spark/SparkPostgresqlJdbc)。
查看Spark写入数据库中的数据
4.1.项目配置源码
4.2.java源码
(完整项目源码Github)
1.首先在postgreSQL中创建一张测试表,并插入数据。(完整项目源码Github)
1.1. 在postgreSQL中的postgres用户下,创建 products
CREATE TABLE products ( product_no integer, name text, price numeric );
1.2. 在 products
插入数据
INSERT INTO products (product_no, name, price) VALUES (1, 'Cheese', 9.99), (2, 'Bread', 1.99), (3, 'Milk', 2.99);
查看数据库写入结果。
2.编写SPARK程序。(完整项目源码Github)
2.1.读取Postgresql某一张表的数据为DataFrame(完整项目源码Github)
SparkPostgresqlJdbc.java
Properties connectionProperties = new Properties(); //增加数据库的用户名(user)密码(password),指定postgresql驱动(driver) connectionProperties.put("user","postgres"); connectionProperties.put("password","123456"); connectionProperties.put("driver","org.postgresql.Driver"); //SparkJdbc读取Postgresql的products表内容 Dataset<Row> jdbcDF = spark.read() .jdbc("jdbc:postgresql://localhost:5432/postgres","products",connectionProperties).select("name","price"); //显示jdbcDF数据内容 jdbcDF.show();
2.2.写入Postgresql某张表中
//将jdbcDF数据新建并写入newproducts,append模式是连接模式,默认的是"error"模式。 jdbcDF.write().mode("append") .jdbc("jdbc:postgresql://localhost:5432/postgres","newproducts",connectionProperties);
3.运行程序,并查看结果(如果在IDEA中开发不熟练,可以看我另一篇博文spark (java API) 在Intellij IDEA中开发并运行)。
3.1.直接在intellij IDEA(社区版)中运行。
a.在运行按钮的“Edit Configeration”中的VM option中添加“-Dspark.master=local”3.2.在终端(Terminal)中运行。
/opt/spark-2.1.0-bin-hadoop2.7/bin/spark-submit \ --class "SparkPostgresqlJdbc" \ --master local[4] \ --driver-class-path /home/xiaolei/.m2/repository/org/postgresql/postgresql/9.4.1212/postgresql-9.4.1212.jar \ target/SparkPostgresqlJdbc-1.0-SNAPSHOT.jar
其中
--driver-class-path指定下载的postgresql JDBC数据
库驱动路径,命令执行要在项目的根目录中(/home/xiaolei/Data/GS/Spark/SparkPostgresqlJdbc)。
查看Spark写入数据库中的数据
4.以下为项目中主要源码(完整项目源码Github):
4.1.项目配置源码pom.xml
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>wangxiaolei</groupId> <artifactId>SparkPostgresqlJdbc</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <!-- Spark dependency --> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>2.1.0</version> </dependency> <dependency> <groupId>org.postgresql</groupId> <artifactId>postgresql</artifactId> <version>9.4.1212</version> </dependency> </dependencies> </project>
4.2.java源码SparkPostgresqlJdbc.java
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import java.util.Properties;
/**
* MIT.
* Author: wangxiaolei(王小雷).
* Date:17-2-9.
* Project:SparkPostgresqlJdbc.
*/
public class SparkPostgresqlJdbc {
public static void main (String[] args) {
SparkSession spark = SparkSession
.builder()
.appName("SparkPostgresqlJdbc")
.config("spark.some.config.option","some-value")
.getOrCreate();
//启动runSparkPostgresqlJdbc程序
runSparkPostgresqlJdbc(spark);
spark.stop();
}
private static void runSparkPostgresqlJdbc(SparkSession spark){
//new一个属性
System.out.println("确保数据库已经开启,并创建了products表和插入了数据");
Properties connectionProperties = new Properties();
//增加数据库的用户名(user)密码(password),指定postgresql驱动(driver)
System.out.println("增加数据库的用户名(user)密码(password),指定postgresql驱动(driver)");
connectionProperties.put("user","postgres");
connectionProperties.put("password","123456");
connectionProperties.put("driver","org.postgresql.Driver");
//SparkJdbc读取Postgresql的products表内容
System.out.println("SparkJdbc读取Postgresql的products表内容");
Dataset<Row> jdbcDF = spark.read()
.jdbc("jdbc:postgresql://localhost:5432/postgres","products",connectionProperties).select("name","price");
//显示jdbcDF数据内容
jdbcDF.show();
//将jdbcDF数据新建并写入newproducts,append模式是连接模式,默认的是"error"模式。 jdbcDF.write().mode("append") .jdbc("jdbc:postgresql://localhost:5432/postgres","newproducts",connectionProperties);
}
}
(完整项目源码Github)
相关文章推荐
- Spark jdbc postgresql数据库连接和写入操作源码解读
- 用java来实现数据库连接查询操作的源码
- Spark jdbc postgresql数据库连接和写入操作源代码解读
- JSP连接操作数据库全接触(所有流行数据库)
- 把数据库连接字符串写入注册表
- asp对数据库的操作中判断连接对象的状态
- ASP常用数据库连接及操作的方法和技巧
- 怎样操作数据库的Blob字段将文件写入数据库!!!
- 用JavaBean封装数据库操作(数据库连接用连接池)- -
- ASP与数据库连接、写入、修改、删除、显示的代码整理
- 总结数据库连接与操作的问题
- SQL Server Mobile 学习(二):通过 VS2005 创建和连接及操作 SQL Server Mobile 数据库
- 注意数据库操作中打开的连接
- 数据库操作异常写入日志文件
- 关于NTKO_office的操作(从数据库中提取数据,写入到NTKO_office_Word中)
- 把数据库连接字符串写入注册表
- 数据库的连接及数据读取,显示操作
- ASP常用数据库连接及操作的方法和技巧
- ASP常用数据库连接及操作的方法和技巧
- 为web.config写入数据库连接字符串的方法