forked from opengaussexamples/examples
!18 添加Spark将openGauss作为数据源的完整示例
Merge pull request !18 from Cerdore/sparkOp
This commit is contained in:
Gitee
commit
fdf0697f87
|
|
@ -0,0 +1,27 @@
|
|||
# SparkOpOpenGauss
|
||||
|
||||
这是将 openGuass 作为数据源的spark示例代码，./src/main/scala/org.openguass.spark.sources.datasourcev2
|
||||
|
||||
环境和软件版本要求（前置要求）
|
||||
1. Java 1.8
|
||||
2. Scala 2.12
|
||||
3. postgresql.jar 或 opengauss-jdbc--${version}.jar(自己打包或官方提供的jar包)
|
||||
|
||||
运行说明：
|
||||
1. 请确保服务器上的数据库正常运行，且你的机器可正常连接数据库。
|
||||
2. 以sparkuser用户身份执行./resources/school.sql文件。
|
||||
3. 请修改代码中sparkuser用户对应的密码、连接数据库的ip及端口，即修改 x.x.x.x:port 。主要代码内容在 `./src/main/scala/`。
|
||||
4. 关于代码的构建
|
||||
+ 首先需要获取 openGauss JDBC 的 jar 包，放入本项目的根目录下的libs目录下,在IDEA中添加jar包为依赖。
|
||||
+ 其次本项目为 maven 项目，IDEA 会通过 maven自动下载其余依赖。
|
||||
+ 如要使用从中央仓库中获取 openGauss JDBC 的 jar 包，可取消掉pom中 openGuass JDBC 的依赖。并将JDBC url 由 `jdbc:postgresql://x.x.x.x:port/dbname` 改为 `jdbc:opengauss://x.x.x.x:port/dbname` ，其中x.x.x.x是IP，port是端口号，dbname是数据库的名字。
|
||||
5. 可以在IDEA中进入以下示例文件中的方法，单击右键可直接运行本示例。
|
||||
+ 可运行`./src/test/scala/org/opengauss/spark/OpenGaussExample`。
|
||||
+ 另一个直接使用Spark JDBC的例子是 `./src/main/scala/SQLDataSourceExample`。
|
||||
|
||||
代码结构说明：
|
||||
1. `./src/main/scala/org.opengauss.spark.sources.datasourcev2`包下分别包含了实现Spark Datasource V2 接口的简单数据源的实现示例、简单多数据源的实现示例、流的实现与测试示例与流和批的实现与测试示例。
|
||||
2. `./src/main/scala/org.opengauss.spark.sources.opengauss`包下包含了将openGauss作为数据源实现Spark Datasouce V2的实现示例。
|
||||
3. `./src/test/scala/org/opengauss/spark/OpenGaussExample`包含了三个测试方法，分别是 简单数据源的使用与测试示例，读和写 openGauss的使用与测试示例。
|
||||
|
||||
PS: Windows 系统在IDEA中调试 `DataSourceV2StreamAndBatchExample` 与 `DataSourceV2StreamingExample`可能会报错，请从[该链接](https://github.com/steveloughran/winutils/tree/master/hadoop-3.0.0/bin) 中下载 hadoop.dll 和 winutils.exe 文件，并将文件路径添加到环境变量中。
|
||||
|
|
@ -0,0 +1,79 @@
|
|||
<?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>org.example</groupId>
|
||||
<artifactId>SparkOpOG</artifactId>
|
||||
<version>3.3.0-SNAPSHOT</version>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<source>8</source>
|
||||
<target>8</target>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.scalatest</groupId>
|
||||
<artifactId>scalatest-maven-plugin</artifactId>
|
||||
<version>1.0</version>
|
||||
<configuration>
|
||||
<reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
|
||||
<junitxml>.</junitxml>
|
||||
<filereports>WDF TestSuite.txt</filereports>
|
||||
</configuration>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>test</id>
|
||||
<goals>
|
||||
<goal>test</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
|
||||
</build>
|
||||
|
||||
<dependencies>
|
||||
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
|
||||
<dependency>
|
||||
<groupId>org.scalatest</groupId>
|
||||
<artifactId>scalatest_2.12</artifactId>
|
||||
<version>3.0.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.spark</groupId>
|
||||
<artifactId>spark-core_2.12</artifactId>
|
||||
<version>3.3.0-SNAPSHOT</version>
|
||||
</dependency>
|
||||
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
|
||||
<dependency>
|
||||
<groupId>org.apache.spark</groupId>
|
||||
<artifactId>spark-sql_2.12</artifactId>
|
||||
<version>3.3.0-SNAPSHOT</version>
|
||||
</dependency>
|
||||
|
||||
<!-- https://mvnrepository.com/artifact/org.opengauss/opengauss-jdbc -->
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>org.opengauss</groupId>-->
|
||||
<!-- <artifactId>opengauss-jdbc</artifactId>-->
|
||||
<!-- <version>2.0.1-compatibility</version>-->
|
||||
<!-- </dependency>-->
|
||||
|
||||
</dependencies>
|
||||
<repositories>
|
||||
<repository>
|
||||
<id>Apache</id>
|
||||
<url>https://repository.apache.org/snapshots/</url>
|
||||
<snapshots>
|
||||
<enabled>true</enabled>
|
||||
</snapshots>
|
||||
</repository>
|
||||
</repositories>
|
||||
</project>
|
||||
|
|
@ -0,0 +1,214 @@
|
|||
create database school;
|
||||
|
||||
\c school;
|
||||
|
||||
BEGIN;
|
||||
|
||||
-- 创建表student
|
||||
CREATE TABLE student
|
||||
(
|
||||
std_id INT PRIMARY KEY,
|
||||
std_name VARCHAR(20) NOT NULL,
|
||||
std_sex VARCHAR(6),
|
||||
std_birth DATE,
|
||||
std_in DATE NOT NULL,
|
||||
std_address VARCHAR(100)
|
||||
);
|
||||
|
||||
-- 创建表teacher
|
||||
CREATE TABLE teacher
|
||||
(
|
||||
tec_id INT PRIMARY KEY,
|
||||
tec_name VARCHAR(20) NOT NULL,
|
||||
tec_job VARCHAR(15),
|
||||
tec_sex VARCHAR(6),
|
||||
tec_age INT,
|
||||
tec_in DATE NOT NULL
|
||||
);
|
||||
|
||||
-- 创建表class
|
||||
CREATE TABLE class
|
||||
(
|
||||
cla_id INT PRIMARY KEY,
|
||||
cla_name VARCHAR(20) NOT NULL,
|
||||
cla_teacher INT NOT NULL
|
||||
);
|
||||
-- 给表class添加外键约束
|
||||
ALTER TABLE class ADD CONSTRAINT fk_tec_id FOREIGN KEY (cla_teacher) REFERENCES teacher(tec_id) ON DELETE CASCADE;
|
||||
|
||||
-- 创建表school_department
|
||||
CREATE TABLE school_department
|
||||
(
|
||||
depart_id INT PRIMARY KEY,
|
||||
depart_name VARCHAR(30) NOT NULL,
|
||||
depart_teacher INT NOT NULL
|
||||
);
|
||||
-- 给表school_department添加外键约束
|
||||
ALTER TABLE school_department ADD CONSTRAINT fk_depart_tec_id FOREIGN KEY (depart_teacher) REFERENCES teacher(tec_id) ON DELETE CASCADE;
|
||||
|
||||
-- 创建表course
|
||||
CREATE TABLE course
|
||||
(
|
||||
cor_id INT PRIMARY KEY,
|
||||
cor_name VARCHAR(30) NOT NULL,
|
||||
cor_type VARCHAR(20),
|
||||
credit DOUBLE PRECISION
|
||||
);
|
||||
|
||||
-- 插入数据
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (1,'张一','男','1993-01-01','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (2,'张二','男','1993-01-02','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (3,'张三','男','1993-01-03','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (4,'张四','男','1993-01-04','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (5,'张五','男','1993-01-05','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (6,'张六','男','1993-01-06','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (7,'张七','男','1993-01-07','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (8,'张八','男','1993-01-08','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (9,'张九','男','1993-01-09','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (10,'李一','男','1993-01-10','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (11,'李二','男','1993-01-11','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (12,'李三','男','1993-01-12','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (13,'李四','男','1993-01-13','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (14,'李五','男','1993-01-14','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (15,'李六','男','1993-01-15','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (16,'李七','男','1993-01-16','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (17,'李八','男','1993-01-17','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (18,'李九','男','1993-01-18','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (19,'王一','男','1993-01-19','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (20,'王二','男','1993-01-20','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (21,'王三','男','1993-01-21','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (22,'王四','男','1993-01-22','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (23,'王五','男','1993-01-23','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (24,'王六','男','1993-01-24','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (25,'王七','男','1993-01-25','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (26,'王八','男','1993-01-26','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (27,'王九','男','1993-01-27','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (28,'钱一','男','1993-01-28','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (29,'钱二','男','1993-01-29','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (30,'钱三','男','1993-01-30','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (31,'钱四','男','1993-02-01','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (32,'钱五','男','1993-02-02','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (33,'钱六','男','1993-02-03','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (34,'钱七','男','1993-02-04','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (35,'钱八','男','1993-02-05','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (36,'钱九','男','1993-02-06','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (37,'吴一','男','1993-02-07','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (38,'吴二','男','1993-02-08','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (39,'吴三','男','1993-02-09','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (40,'吴四','男','1993-02-10','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (41,'吴五','男','1993-02-11','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (42,'吴六','男','1993-02-12','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (43,'吴七','男','1993-02-13','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (44,'吴八','男','1993-02-14','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (45,'吴九','男','1993-02-15','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (46,'柳一','男','1993-02-16','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (47,'柳二','男','1993-02-17','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (48,'柳三','男','1993-02-18','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (49,'柳四','男','1993-02-19','2011-09-01','江苏省南京市雨花台区');
|
||||
INSERT INTO student(std_id,std_name,std_sex,std_birth,std_in,std_address) VALUES (50,'柳五','男','1993-02-20','2011-09-01','江苏省南京市雨花台区');
|
||||
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (1,'张一','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (2,'张二','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (3,'张三','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (4,'张四','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (5,'张五','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (6,'张六','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (7,'张七','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (8,'张八','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (9,'张九','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (10,'李一','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (11,'李二','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (12,'李三','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (13,'李四','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (14,'李五','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (15,'李六','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (16,'李七','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (17,'李八','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (18,'李九','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (19,'王一','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (20,'王二','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (21,'王三','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (22,'王四','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (23,'王五','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (24,'王六','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (25,'王七','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (26,'王八','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (27,'王九','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (28,'钱一','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (29,'钱二','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (30,'钱三','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (31,'钱四','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (32,'钱五','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (33,'钱六','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (34,'钱七','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (35,'钱八','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (36,'钱九','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (37,'吴一','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (38,'吴二','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (39,'吴三','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (40,'吴四','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (41,'吴五','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (42,'吴六','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (43,'吴七','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (44,'吴八','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (45,'吴九','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (46,'柳一','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (47,'柳二','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (48,'柳三','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (49,'柳四','讲师','男',35,'2009-07-01');
|
||||
INSERT INTO teacher(tec_id,tec_name,tec_job,tec_sex,tec_age,tec_in) VALUES (50,'柳五','讲师','男',35,'2009-07-01');
|
||||
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (1,'计算机',1);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (2,'自动化',3);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (3,'飞行器设计',5);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (4,'大学物理',7);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (5,'高等数学',9);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (6,'大学化学',12);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (7,'表演',14);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (8,'服装设计',16);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (9,'工业设计',18);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (10,'金融学',21);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (11,'医学',23);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (12,'土木工程',25);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (13,'机械',27);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (14,'建筑学',29);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (15,'经济学',32);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (16,'财务管理',34);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (17,'人力资源',36);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (18,'力学',38);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (19,'人工智能',41);
|
||||
INSERT INTO class(cla_id,cla_name,cla_teacher) VALUES (20,'会计',45);
|
||||
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (1,'计算机学院',2);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (2,'自动化学院',4);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (3,'航空宇航学院',6);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (4,'艺术学院',8);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (5,'理学院',11);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (6,'人工智能学院',13);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (7,'工学院',15);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (8,'管理学院',17);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (9,'农学院',22);
|
||||
INSERT INTO school_department(depart_id,depart_name,depart_teacher) VALUES (10,'医学院',28);
|
||||
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (1,'数据库系统概论','必修',3);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (2,'艺术设计概论','选修',1);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (3,'力学制图','必修',4);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (4,'飞行器设计历史','选修',1);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (5,'马克思主义','必修',2);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (6,'大学历史','必修',2);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (7,'人力资源管理理论','必修',2.5);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (8,'线性代数','必修',4);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (9,'JAVA程序设计','必修',3);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (10,'操作系统','必修',4);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (11,'计算机组成原理','必修',3);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (12,'自动化设计理论','必修',2);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (13,'情绪表演','必修',2.5);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (14,'茶学历史','选修',1);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (15,'艺术论','必修',1.5);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (16,'机器学习','必修',3);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (17,'数据挖掘','选修',2);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (18,'图像识别','必修',3);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (19,'解剖学','必修',4);
|
||||
INSERT INTO course(cor_id,cor_name,cor_type,credit) VALUES (20,'3D max','选修',2);
|
||||
|
||||
COMMIT;
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
package org.apache.spark.examples.sql
|
||||
|
||||
import java.util.Properties
|
||||
|
||||
import org.apache.spark.sql.SparkSession
|
||||
|
||||
object SQLDataSourceExample {
|
||||
|
||||
case class Person(name: String, age: Long)
|
||||
|
||||
def main(args: Array[String]): Unit = {
|
||||
val spark = SparkSession
|
||||
.builder()
|
||||
.master("local")
|
||||
.appName("Spark SQL data sources example")
|
||||
.config("spark.some.config.option", "some-value")
|
||||
.getOrCreate()
|
||||
|
||||
runJdbcDatasetExample(spark)
|
||||
|
||||
spark.stop()
|
||||
}
|
||||
|
||||
|
||||
private def runJdbcDatasetExample(spark: SparkSession): Unit = {
|
||||
// $example on:jdbc_dataset$
|
||||
// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
|
||||
// Loading data from a JDBC source
|
||||
val dburl = "jdbc:postgresql://x.x.x.x:port/school" //注意修改此处的ip与端口地址
|
||||
val jdbcDF = spark.read
|
||||
.format("jdbc")
|
||||
.option("url", dburl)
|
||||
.option("dbtable", "class")
|
||||
.option("user", "sparkuser")
|
||||
.option("password", "pwdofsparkuser")
|
||||
.load()
|
||||
|
||||
val connectionProperties = new Properties()
|
||||
connectionProperties.put("user", "sparkuser")
|
||||
connectionProperties.put("password", "pwdofsparkuser")
|
||||
val jdbcDF2 = spark.read
|
||||
.option("customSchema","cla_id INT, cla_teacher STRING")
|
||||
.jdbc(dburl, "class", connectionProperties)
|
||||
// Specifying the custom data types of the read schema
|
||||
|
||||
|
||||
connectionProperties.put("customSchema", "cla_id INT, cla_name STRING")
|
||||
val jdbcDF3 = spark.read
|
||||
.jdbc(dburl, "class", connectionProperties)
|
||||
|
||||
// Saving data to a JDBC source. Create table "customtable1", and write data
|
||||
jdbcDF.write
|
||||
.format("jdbc")
|
||||
.option("url", dburl)
|
||||
.option("dbtable", "customtable1")
|
||||
.option("user", "sparkuser")
|
||||
.option("password", "pwdofsparkuser")
|
||||
.save()
|
||||
|
||||
|
||||
jdbcDF2.write
|
||||
.jdbc(dburl, "customtable2", connectionProperties)
|
||||
|
||||
// Specifying create table column data types on write
|
||||
jdbcDF3.write
|
||||
.option("createTableColumnTypes", "cla_id INT, cla_name VARCHAR(20)")
|
||||
.jdbc(dburl, "customtable3", connectionProperties)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,103 @@
|
|||
package org.opengauss.spark.sources.datasourcev2.multi
|
||||
|
||||
import java.util
|
||||
|
||||
import org.apache.spark.sql.catalyst.InternalRow
|
||||
import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability, TableProvider}
|
||||
import org.apache.spark.sql.connector.expressions.Transform
|
||||
import org.apache.spark.sql.connector.read._
|
||||
import org.apache.spark.sql.types.{StringType, StructField, StructType}
|
||||
import org.apache.spark.sql.util.CaseInsensitiveStringMap
|
||||
import org.apache.spark.unsafe.types.UTF8String
|
||||
|
||||
import scala.collection.JavaConverters._
|
||||
|
||||
/*
|
||||
* Default source should some kind of relation provider
|
||||
*/
|
||||
class DefaultSource extends TableProvider {
|
||||
|
||||
override def inferSchema(caseInsensitiveStringMap: CaseInsensitiveStringMap): StructType =
|
||||
getTable(null, Array.empty[Transform], caseInsensitiveStringMap.asCaseSensitiveMap()).schema()
|
||||
|
||||
override def getTable(structType: StructType, transforms: Array[Transform], map: util.Map[String, String]): Table =
|
||||
new SimpleBatchTable()
|
||||
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
Defines Read Support and Initial Schema
|
||||
*/
|
||||
|
||||
class SimpleBatchTable extends Table with SupportsRead {
|
||||
override def name(): String = this.getClass.toString
|
||||
|
||||
override def schema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def capabilities(): util.Set[TableCapability] = Set(TableCapability.BATCH_READ).asJava
|
||||
|
||||
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = new SimpleScanBuilder()
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
Scan object with no mixins
|
||||
*/
|
||||
class SimpleScanBuilder extends ScanBuilder {
|
||||
override def build(): Scan = new SimpleScan
|
||||
}
|
||||
|
||||
/*
|
||||
Batch Reading Support
|
||||
|
||||
The schema is repeated here as it can change after column pruning etc
|
||||
*/
|
||||
|
||||
class SimpleScan extends Scan with Batch {
|
||||
override def readSchema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def toBatch: Batch = this
|
||||
|
||||
override def planInputPartitions(): Array[InputPartition] = {
|
||||
Array(new SimplePartition(0, 4),
|
||||
new SimplePartition(5, 9))
|
||||
}
|
||||
|
||||
override def createReaderFactory(): PartitionReaderFactory = new SimplePartitionReaderFactory()
|
||||
}
|
||||
|
||||
// simple class to organise the partition
|
||||
class SimplePartition(val start: Int, val end: Int) extends InputPartition
|
||||
|
||||
// reader factory
|
||||
class SimplePartitionReaderFactory extends PartitionReaderFactory {
|
||||
override def createReader(partition: InputPartition): PartitionReader[InternalRow] = new
|
||||
SimplePartitionReader(partition.asInstanceOf[SimplePartition])
|
||||
}
|
||||
|
||||
|
||||
// parathion reader
|
||||
class SimplePartitionReader(inputPartition: SimplePartition) extends PartitionReader[InternalRow] {
|
||||
|
||||
val values = Array("1", "2", "3", "4", "5", "6", "7", "8", "9", "10")
|
||||
|
||||
var index = inputPartition.start
|
||||
|
||||
def next = index <= inputPartition.end
|
||||
|
||||
def get = {
|
||||
val stringValue = values(index)
|
||||
val stringUtf = UTF8String.fromString(stringValue)
|
||||
val row = InternalRow(stringUtf)
|
||||
index = index + 1
|
||||
row
|
||||
}
|
||||
|
||||
def close() = Unit
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,100 @@
|
|||
package org.opengauss.spark.sources.datasourcev2.simple
|
||||
|
||||
import java.util
|
||||
|
||||
import org.apache.spark.sql.catalyst.InternalRow
|
||||
import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability, TableProvider}
|
||||
import org.apache.spark.sql.connector.expressions.Transform
|
||||
import org.apache.spark.sql.connector.read._
|
||||
import org.apache.spark.sql.types.{StringType, StructField, StructType}
|
||||
import org.apache.spark.sql.util.CaseInsensitiveStringMap
|
||||
import org.apache.spark.unsafe.types.UTF8String
|
||||
|
||||
import scala.collection.JavaConverters._
|
||||
|
||||
/*
|
||||
* Default source should some kind of relation provider
|
||||
*/
|
||||
class DefaultSource extends TableProvider{
|
||||
|
||||
override def inferSchema(caseInsensitiveStringMap: CaseInsensitiveStringMap): StructType =
|
||||
getTable(null,Array.empty[Transform],caseInsensitiveStringMap.asCaseSensitiveMap()).schema()
|
||||
|
||||
override def getTable(structType: StructType, transforms: Array[Transform], map: util.Map[String, String]): Table =
|
||||
new SimpleBatchTable()
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
Defines Read Support and Initial Schema
|
||||
*/
|
||||
|
||||
class SimpleBatchTable extends Table with SupportsRead {
|
||||
override def name(): String = this.getClass.toString
|
||||
|
||||
override def schema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def capabilities(): util.Set[TableCapability] = Set(TableCapability.BATCH_READ).asJava
|
||||
|
||||
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = new SimpleScanBuilder()
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*
|
||||
Scan object with no mixins
|
||||
*/
|
||||
class SimpleScanBuilder extends ScanBuilder {
|
||||
override def build(): Scan = new SimpleScan
|
||||
}
|
||||
|
||||
/*
|
||||
Batch Reading Support
|
||||
|
||||
The schema is repeated here as it can change after column pruning etc
|
||||
*/
|
||||
|
||||
class SimpleScan extends Scan with Batch{
|
||||
override def readSchema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def toBatch: Batch = this
|
||||
|
||||
override def planInputPartitions(): Array[InputPartition] = {
|
||||
Array(new SimplePartition())
|
||||
}
|
||||
override def createReaderFactory(): PartitionReaderFactory = new SimplePartitionReaderFactory()
|
||||
}
|
||||
|
||||
// simple class to organise the partition
|
||||
class SimplePartition extends InputPartition
|
||||
|
||||
// reader factory
|
||||
class SimplePartitionReaderFactory extends PartitionReaderFactory {
|
||||
override def createReader(partition: InputPartition): PartitionReader[InternalRow] = new SimplePartitionReader
|
||||
}
|
||||
|
||||
|
||||
// parathion reader
|
||||
class SimplePartitionReader extends PartitionReader[InternalRow] {
|
||||
|
||||
val values = Array("1", "2", "3", "4", "5")
|
||||
|
||||
var index = 0
|
||||
|
||||
def next = index < values.length
|
||||
|
||||
def get = {
|
||||
val stringValue = values(index)
|
||||
val stringUtf = UTF8String.fromString(stringValue)
|
||||
val row = InternalRow(stringUtf)
|
||||
index = index + 1
|
||||
row
|
||||
}
|
||||
|
||||
def close() = Unit
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
package org.opengauss.spark.sources.datasourcev2.streamandbatch
|
||||
|
||||
import org.apache.spark.sql.SparkSession
|
||||
import org.apache.spark.sql.streaming.OutputMode
|
||||
|
||||
object DataSourceV2StreamAndBatchExample {
|
||||
def main(args: Array[String]): Unit = {
|
||||
val sparkSession = SparkSession.builder.
|
||||
master("local[2]")
|
||||
.appName("streaming example")
|
||||
.getOrCreate()
|
||||
|
||||
|
||||
val dataSource = "org.opengauss.spark.sources.datasourcev2.streamandbatch"
|
||||
|
||||
val batchDf = sparkSession
|
||||
.read
|
||||
.format(dataSource)
|
||||
.load()
|
||||
|
||||
batchDf.show()
|
||||
|
||||
val streamingDf = sparkSession.
|
||||
readStream.
|
||||
format(dataSource)
|
||||
.load()
|
||||
|
||||
val query = streamingDf.writeStream
|
||||
.format("console")
|
||||
.queryName("simple_source")
|
||||
.outputMode(OutputMode.Append())
|
||||
|
||||
query.start().awaitTermination()
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,130 @@
|
|||
package org.opengauss.spark.sources.datasourcev2.streamandbatch
|
||||
|
||||
import java.util
|
||||
|
||||
import org.apache.spark.sql.catalyst.InternalRow
|
||||
import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability, TableProvider}
|
||||
import org.apache.spark.sql.connector.expressions.Transform
|
||||
import org.apache.spark.sql.connector.read._
|
||||
import org.apache.spark.sql.connector.read.streaming.{MicroBatchStream, Offset}
|
||||
import org.apache.spark.sql.types.{StringType, StructField, StructType}
|
||||
import org.apache.spark.sql.util.CaseInsensitiveStringMap
|
||||
import org.apache.spark.unsafe.types.UTF8String
|
||||
|
||||
import scala.collection.JavaConverters._
|
||||
|
||||
/*
|
||||
* Default source should some kind of relation provider
|
||||
*/
|
||||
class DefaultSource extends TableProvider{
|
||||
|
||||
override def inferSchema(caseInsensitiveStringMap: CaseInsensitiveStringMap): StructType =
|
||||
getTable(null,Array.empty[Transform],caseInsensitiveStringMap.asCaseSensitiveMap()).schema()
|
||||
|
||||
override def getTable(structType: StructType, transforms: Array[Transform], map: util.Map[String, String]): Table =
|
||||
new SimpleStreamingTable()
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
Defines Read Support and Initial Schema
|
||||
*/
|
||||
|
||||
class SimpleStreamingTable extends Table with SupportsRead {
|
||||
override def name(): String = this.getClass.toString
|
||||
|
||||
override def schema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def capabilities(): util.Set[TableCapability] = Set(TableCapability.MICRO_BATCH_READ,
|
||||
TableCapability.BATCH_READ).asJava
|
||||
|
||||
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = new SimpleScanBuilder()
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
Scan object with no mixins
|
||||
*/
|
||||
class SimpleScanBuilder extends ScanBuilder {
|
||||
override def build(): Scan = new SimpleScan
|
||||
}
|
||||
|
||||
/*
|
||||
Batch Reading Support
|
||||
|
||||
The schema is repeated here as it can change after column pruning etc
|
||||
*/
|
||||
|
||||
class SimpleScan extends Scan{
|
||||
override def readSchema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream = new SimpleMicroBatchStream()
|
||||
|
||||
override def toBatch: Batch = new SimpleBatch
|
||||
}
|
||||
|
||||
class SimpleBatch extends Batch{
|
||||
override def planInputPartitions(): Array[InputPartition] = Array(new SimplePartition)
|
||||
|
||||
override def createReaderFactory(): PartitionReaderFactory = new SimplePartitionReaderFactory
|
||||
}
|
||||
|
||||
class SimpleOffset(value:Int) extends Offset {
|
||||
override def json(): String = s"""{"value":"$value"}"""
|
||||
}
|
||||
|
||||
class SimpleMicroBatchStream extends MicroBatchStream {
|
||||
var latestOffsetValue = 0
|
||||
|
||||
override def latestOffset(): Offset = {
|
||||
latestOffsetValue += 10
|
||||
new SimpleOffset(latestOffsetValue)
|
||||
}
|
||||
|
||||
override def planInputPartitions(offset: Offset, offset1: Offset): Array[InputPartition] = Array(new SimplePartition)
|
||||
|
||||
override def createReaderFactory(): PartitionReaderFactory = new SimplePartitionReaderFactory()
|
||||
|
||||
override def initialOffset(): Offset = new SimpleOffset(latestOffsetValue)
|
||||
|
||||
override def deserializeOffset(s: String): Offset = new SimpleOffset(latestOffsetValue)
|
||||
|
||||
override def commit(offset: Offset): Unit = {}
|
||||
|
||||
override def stop(): Unit = {}
|
||||
}
|
||||
|
||||
|
||||
// simple class to organise the partition
|
||||
class SimplePartition extends InputPartition
|
||||
|
||||
// reader factory
|
||||
class SimplePartitionReaderFactory extends PartitionReaderFactory {
|
||||
override def createReader(partition: InputPartition): PartitionReader[InternalRow] = new SimplePartitionReader
|
||||
}
|
||||
|
||||
|
||||
// parathion reader
|
||||
class SimplePartitionReader extends PartitionReader[InternalRow] {
|
||||
|
||||
val values = Array("1", "2", "3", "4", "5")
|
||||
|
||||
var index = 0
|
||||
|
||||
def next = index < values.length
|
||||
|
||||
def get = {
|
||||
val stringValue = values(index)
|
||||
val stringUtf = UTF8String.fromString(stringValue)
|
||||
val row = InternalRow(stringUtf)
|
||||
index = index + 1
|
||||
row
|
||||
}
|
||||
|
||||
def close() = Unit
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
package org.opengauss.spark.sources.datasourcev2.streaming
|
||||
|
||||
import org.apache.spark.sql.SparkSession
|
||||
import org.apache.spark.sql.streaming.OutputMode
|
||||
|
||||
object DataSourceV2StreamingExample {
|
||||
def main(args: Array[String]): Unit = {
|
||||
val sparkSession = SparkSession.builder.
|
||||
master("local[2]")
|
||||
.appName("streaming example")
|
||||
.getOrCreate()
|
||||
|
||||
val streamingDf = sparkSession.
|
||||
readStream.
|
||||
format("org.opengauss.spark.sources.datasourcev2.streaming")
|
||||
.load()
|
||||
|
||||
|
||||
|
||||
val query = streamingDf.writeStream
|
||||
.format("console")
|
||||
.queryName("simple_source")
|
||||
.outputMode(OutputMode.Append())
|
||||
|
||||
query.start().awaitTermination()
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,121 @@
|
|||
package org.opengauss.spark.sources.datasourcev2.streaming
|
||||
|
||||
import java.util
|
||||
|
||||
import org.apache.spark.sql.catalyst.InternalRow
|
||||
import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability, TableProvider}
|
||||
import org.apache.spark.sql.connector.expressions.Transform
|
||||
import org.apache.spark.sql.connector.read._
|
||||
import org.apache.spark.sql.connector.read.streaming.{MicroBatchStream, Offset}
|
||||
import org.apache.spark.sql.types.{StringType, StructField, StructType}
|
||||
import org.apache.spark.sql.util.CaseInsensitiveStringMap
|
||||
import org.apache.spark.unsafe.types.UTF8String
|
||||
|
||||
import scala.collection.JavaConverters._
|
||||
|
||||
/*
|
||||
* Default source should some kind of relation provider
|
||||
*/
|
||||
class DefaultSource extends TableProvider{
|
||||
|
||||
override def inferSchema(caseInsensitiveStringMap: CaseInsensitiveStringMap): StructType =
|
||||
getTable(null,Array.empty[Transform],caseInsensitiveStringMap.asCaseSensitiveMap()).schema()
|
||||
|
||||
override def getTable(structType: StructType, transforms: Array[Transform], map: util.Map[String, String]): Table =
|
||||
new SimpleStreamingTable()
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
Defines Read Support and Initial Schema
|
||||
*/
|
||||
|
||||
class SimpleStreamingTable extends Table with SupportsRead {
|
||||
override def name(): String = this.getClass.toString
|
||||
|
||||
override def schema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def capabilities(): util.Set[TableCapability] = Set(TableCapability.MICRO_BATCH_READ).asJava
|
||||
|
||||
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = new SimpleScanBuilder()
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
Scan object with no mixins
|
||||
*/
|
||||
class SimpleScanBuilder extends ScanBuilder {
|
||||
override def build(): Scan = new SimpleScan
|
||||
}
|
||||
|
||||
/*
|
||||
Batch Reading Support
|
||||
|
||||
The schema is repeated here as it can change after column pruning etc
|
||||
*/
|
||||
|
||||
class SimpleScan extends Scan{
|
||||
override def readSchema(): StructType = StructType(Array(StructField("value", StringType)))
|
||||
|
||||
override def toMicroBatchStream(checkpointLocation: String): MicroBatchStream = new SimpleMicroBatchStream()
|
||||
}
|
||||
|
||||
class SimpleOffset(value:Int) extends Offset {
|
||||
override def json(): String = s"""{"value":"$value"}"""
|
||||
}
|
||||
|
||||
class SimpleMicroBatchStream extends MicroBatchStream {
|
||||
var latestOffsetValue = 0
|
||||
|
||||
override def latestOffset(): Offset = {
|
||||
latestOffsetValue += 10
|
||||
new SimpleOffset(latestOffsetValue)
|
||||
}
|
||||
|
||||
override def planInputPartitions(offset: Offset, offset1: Offset): Array[InputPartition] = Array(new SimplePartition)
|
||||
|
||||
override def createReaderFactory(): PartitionReaderFactory = new SimplePartitionReaderFactory()
|
||||
|
||||
override def initialOffset(): Offset = new SimpleOffset(latestOffsetValue)
|
||||
|
||||
override def deserializeOffset(s: String): Offset = new SimpleOffset(latestOffsetValue)
|
||||
|
||||
override def commit(offset: Offset): Unit = {}
|
||||
|
||||
override def stop(): Unit = {}
|
||||
}
|
||||
|
||||
|
||||
// simple class to organise the partition
|
||||
class SimplePartition extends InputPartition
|
||||
|
||||
// reader factory
|
||||
class SimplePartitionReaderFactory extends PartitionReaderFactory {
|
||||
override def createReader(partition: InputPartition): PartitionReader[InternalRow] = new SimplePartitionReader
|
||||
}
|
||||
|
||||
|
||||
// parathion reader
|
||||
class SimplePartitionReader extends PartitionReader[InternalRow] {
|
||||
|
||||
val values = Array("1", "2", "3", "4", "5")
|
||||
|
||||
var index = 0
|
||||
|
||||
def next = index < values.length
|
||||
|
||||
def get = {
|
||||
val stringValue = values(index)
|
||||
val stringUtf = UTF8String.fromString(stringValue)
|
||||
val row = InternalRow(stringUtf)
|
||||
index = index + 1
|
||||
row
|
||||
}
|
||||
|
||||
def close() = Unit
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,151 @@
|
|||
package org.opengauss.spark.sources.opengauss
|
||||
|
||||
import java.sql.DriverManager
|
||||
import java.util
|
||||
|
||||
import org.apache.spark.sql.catalyst.InternalRow
|
||||
import org.apache.spark.sql.connector.catalog._
|
||||
import org.apache.spark.sql.connector.expressions.Transform
|
||||
import org.apache.spark.sql.connector.read._
|
||||
import org.apache.spark.sql.connector.write._
|
||||
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}
|
||||
import org.apache.spark.sql.util.CaseInsensitiveStringMap
|
||||
import org.apache.spark.unsafe.types.UTF8String
|
||||
|
||||
import scala.collection.JavaConverters._
|
||||
|
||||
class DefaultSource extends TableProvider {
|
||||
override def inferSchema(options: CaseInsensitiveStringMap): StructType = OpenGaussTable.schema
|
||||
|
||||
override def getTable(
|
||||
schema: StructType,
|
||||
partitioning: Array[Transform],
|
||||
properties: util.Map[String, String]
|
||||
): Table = new OpenGaussTable(properties.get("tableName")) // TODO: Error handling
|
||||
}
|
||||
|
||||
class OpenGaussTable(val name: String) extends SupportsRead with SupportsWrite {
|
||||
override def schema(): StructType = OpenGaussTable.schema
|
||||
|
||||
override def capabilities(): util.Set[TableCapability] = Set(
|
||||
TableCapability.BATCH_READ,
|
||||
TableCapability.BATCH_WRITE
|
||||
).asJava
|
||||
|
||||
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = new OpenGaussScanBuilder(options)
|
||||
|
||||
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = new OpenGaussWriteBuilder(info.options)
|
||||
}
|
||||
|
||||
object OpenGaussTable {
|
||||
/*Table products*/
|
||||
/*Database school, table course*/
|
||||
val schema: StructType = new StructType().add("cor_id", IntegerType).add("cor_name", StringType).add("cor_type", StringType).add("credit", DoubleType)
|
||||
}
|
||||
|
||||
case class ConnectionProperties(url: String, user: String, password: String, tableName: String, partitionColumn: String, partitionSize: Int)
|
||||
|
||||
|
||||
|
||||
/** Read */
|
||||
|
||||
class OpenGaussScanBuilder(options: CaseInsensitiveStringMap) extends ScanBuilder {
|
||||
override def build(): Scan = new OpenGaussScan(ConnectionProperties(
|
||||
options.get("url"), options.get("user"), options.get("password"), options.get("tableName"), options.get("partitionColumn"), options.get("partitionSize").toInt
|
||||
))
|
||||
}
|
||||
|
||||
class OpenGaussPartition extends InputPartition
|
||||
|
||||
class OpenGaussScan(connectionProperties: ConnectionProperties) extends Scan with Batch {
|
||||
override def readSchema(): StructType = OpenGaussTable.schema
|
||||
|
||||
override def toBatch: Batch = this
|
||||
|
||||
override def planInputPartitions(): Array[InputPartition] = Array(new OpenGaussPartition)
|
||||
|
||||
override def createReaderFactory(): PartitionReaderFactory = new OpenGaussPartitionReaderFactory(connectionProperties)
|
||||
}
|
||||
|
||||
class OpenGaussPartitionReaderFactory(connectionProperties: ConnectionProperties)
|
||||
extends PartitionReaderFactory {
|
||||
override def createReader(partition: InputPartition): PartitionReader[InternalRow] = new OpenGaussPartitionReader(connectionProperties)
|
||||
}
|
||||
|
||||
|
||||
|
||||
class OpenGaussPartitionReader(connectionProperties: ConnectionProperties) extends PartitionReader[InternalRow] {
|
||||
private val connection = DriverManager.getConnection(
|
||||
connectionProperties.url, connectionProperties.user, connectionProperties.password
|
||||
)
|
||||
private val statement = connection.createStatement()
|
||||
private val resultSet = statement.executeQuery(s"select * from ${connectionProperties.tableName}")
|
||||
|
||||
override def next(): Boolean = resultSet.next()
|
||||
|
||||
override def get(): InternalRow = InternalRow(
|
||||
resultSet.getInt(1),
|
||||
UTF8String.fromString(resultSet.getString(2)),
|
||||
UTF8String.fromString(resultSet.getString(3)),
|
||||
resultSet.getDouble(4))
|
||||
|
||||
override def close(): Unit = connection.close()
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
/** Write */
|
||||
|
||||
class OpenGaussWriteBuilder(options: CaseInsensitiveStringMap) extends WriteBuilder {
|
||||
override def buildForBatch(): BatchWrite = new OpenGaussBatchWrite(ConnectionProperties(
|
||||
options.get("url"), options.get("user"), options.get("password"), options.get("tableName"), options.get("partitionColumn"), options.get("partitionSize").toInt
|
||||
))
|
||||
}
|
||||
|
||||
class OpenGaussBatchWrite(connectionProperties: ConnectionProperties) extends BatchWrite {
|
||||
override def createBatchWriterFactory(physicalWriteInfo: PhysicalWriteInfo): DataWriterFactory =
|
||||
new OpenGaussDataWriterFactory(connectionProperties)
|
||||
|
||||
override def commit(writerCommitMessages: Array[WriterCommitMessage]): Unit = {}
|
||||
|
||||
override def abort(writerCommitMessages: Array[WriterCommitMessage]): Unit = {}
|
||||
}
|
||||
|
||||
class OpenGaussDataWriterFactory(connectionProperties: ConnectionProperties) extends DataWriterFactory {
|
||||
override def createWriter(partitionId: Int, taskId:Long): DataWriter[InternalRow] =
|
||||
new OpenGaussWriter(connectionProperties)
|
||||
}
|
||||
|
||||
object WriteSucceeded extends WriterCommitMessage
|
||||
|
||||
class OpenGaussWriter(connectionProperties: ConnectionProperties) extends DataWriter[InternalRow] {
|
||||
|
||||
val connection = DriverManager.getConnection(
|
||||
connectionProperties.url,
|
||||
connectionProperties.user,
|
||||
connectionProperties.password
|
||||
)
|
||||
|
||||
val statement = "insert into ${connectionProperties.tableName} (cla_id, cla_name, cla_teacher) values (?,?,?)"
|
||||
val preparedStatement = connection.prepareStatement(statement)
|
||||
|
||||
override def write(record: InternalRow): Unit = {
|
||||
val cla_id = record.getInt(0)
|
||||
val cla_name = record.getString(1)
|
||||
val cla_teacher = record.getInt(2)
|
||||
|
||||
preparedStatement.setInt(0, cla_id)
|
||||
preparedStatement.setString(1, cla_name)
|
||||
preparedStatement.setInt(2, cla_teacher)
|
||||
preparedStatement.executeUpdate()
|
||||
}
|
||||
|
||||
override def commit(): WriterCommitMessage = WriteSucceeded
|
||||
|
||||
override def abort(): Unit = {}
|
||||
|
||||
override def close(): Unit = connection.close()
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,79 @@
|
|||
package org.opengauss.spark
|
||||
|
||||
import org.apache.spark.sql.{SaveMode, SparkSession}
|
||||
import org.scalatest.FlatSpec
|
||||
import java.sql.DriverManager
|
||||
import java.util.Properties
|
||||
|
||||
import org.scalatest.Matchers.convertToAnyShouldWrapper
|
||||
|
||||
class OpenGaussExample extends FlatSpec {
|
||||
|
||||
val testTableName = "course"
|
||||
val dburl = "jdbc:postgresql://x.x.x.x:port/school" //注意将此处修改成你的机器对应的的ip与端口
|
||||
|
||||
"Simple data source" should "read" in{
|
||||
val sparkSession = SparkSession.builder
|
||||
.master("local[2]")
|
||||
.appName("example")
|
||||
.getOrCreate()
|
||||
|
||||
val simpleDf = sparkSession.read
|
||||
.format("org.opengauss.spark.sources.datasourcev2.simple")
|
||||
.load()
|
||||
|
||||
simpleDf.show()
|
||||
println(
|
||||
"number of partitions in simple source is " + simpleDf.rdd.getNumPartitions)
|
||||
}
|
||||
|
||||
|
||||
"openGauss data source" should "read table" in {
|
||||
val spark = SparkSession
|
||||
.builder()
|
||||
.master("local[*]")
|
||||
.appName("OpenGaussReaderJob")
|
||||
.getOrCreate()
|
||||
|
||||
val simpleRead = spark
|
||||
.read
|
||||
.format("org.opengauss.spark.sources.opengauss")
|
||||
.option("url", dburl)
|
||||
.option("user", "sparkuser")
|
||||
.option("password", "pwdofsparkuser")
|
||||
.option("tableName", testTableName)
|
||||
.option("partitionSize", 10)
|
||||
.load()
|
||||
.show()
|
||||
|
||||
spark.stop()
|
||||
}
|
||||
|
||||
"openGauss data source" should "write table" in {
|
||||
val spark = SparkSession
|
||||
.builder()
|
||||
.master("local[*]")
|
||||
.appName("OpenGaussWriterJob")
|
||||
.getOrCreate()
|
||||
|
||||
import spark.implicits._
|
||||
|
||||
val df = (60 to 70).map(_.toLong).toDF("cla_id")
|
||||
|
||||
df
|
||||
.write
|
||||
.format("org.opengauss.spark.sources.opengauss")
|
||||
.option("url", dburl)
|
||||
.option("user", "sparkuser")
|
||||
.option("password", "pwdofsparkuser")
|
||||
.option("tableName", testTableName)
|
||||
.option("partitionSize", 10)
|
||||
.option("partitionColumn", "cla_id")
|
||||
.mode(SaveMode.Append)
|
||||
.save()
|
||||
|
||||
spark.stop()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue