JDBC驱动

本文档将介绍 SparkSQL 通过 JDBC 驱动对接 SequoiaDB 巨杉数据库的示例。

SparkSQL连接SequoiaDB

SparkSQL 可以通过 JDBC 驱动连接 SequoiaDB 进行操作。

连接前准备

  1. 下载安装 Spark 和 SequoiaDB 数据库，将 Spark-SequoiaDB 连接组件和 SequoiaDB Java 驱动的 jar 包复制到 Spark 安装路径下的 jars 目录下

  2. 新建一个 java 项目，并导入 sparkSQL 的 JDBC 驱动程序依赖包，可使用 maven 下载，参考配置如下：

    1.  <dependencies>
    2.    <dependency>
    3.         <groupId>org.apache.hive</groupId>
    4.         <artifactId>hive-jdbc</artifactId>
    5.         <version>$version</version>
    6.    </dependency>
    7.    <dependency>
    8.         <groupId>org.apache.hadoop</groupId>
    9.         <artifactId>hadoop-common</artifactId>   
    10.         <version>$version</version>     
    11.     </dependency>
    12.   </dependencies>

示例

假设 SequoiaDB 存在集合 test.test，且保存数据如下：

  1. > db.test.test.find()
  2. {
  3.   "_id": {
  4.     "$oid": "5d5911f41125bc9c9aa2bc0b"
  5.   },
  6.   "c1": 0,
  7.   "c2": "mary",
  8.   "c3": 15
  9. }
  10. {
  11.   "_id": {
  12.     "$oid": "5d5912041125bc9c9aa2bc0c"
  13.   },
  14.   "c1": 1,
  15.   "c2": "lili",
  16.   "c3": 25
  17. }

编写并执行示例代码

  1. package com.spark.samples;
  3. import java.sql.Connection;
  4. import java.sql.DriverManager;
  5. import java.sql.ResultSet;
  6. import java.sql.SQLException;
  7. import java.sql.Statement;
  9. public class HiveJdbcClient {
  10.     public static void main(String[] args) throws ClassNotFoundException {
  11.         //JDBC Driver程序的类名
  12.         Class.forName("org.apache.hive.jdbc.HiveDriver");
  13.         try {
  14.             //连接SparkSQL，假设spark服务所在主机名为sparkServer
  15.             Connection connection = DriverManager.getConnection("jdbc:hive2://sparkServer:10000/default", "", "");
  16.             System.out.println("connection success!");
  17.             Statement statement = connection.createStatement();
  18.             // 创建表，该表映射SequoiaDB中表test.test
  19.             String crtTableName = "test";
  20.             statement.execute("CREATE TABLE" + crtTableName
  21.                     + "( c1 int, c2 string, c3 int ) USING com.sequoiadb.spark OPTIONS ( host 'server1:11810,server2:11810', "
  22.                     + "collectionspace 'test', collection 'test',username '',password '')");
  23.             // 查询表test数据,返回sequoiaDB中test.test表中的数据信息
  24.             String sql = "select * from " + crtTableName;
  25.             System.out.println("Running:" + sql);
  26.             ResultSet resultSet = statement.executeQuery(sql);
  27.             while (resultSet.next()) {
  28.                 System.out.println(
  29.                         String.valueOf(resultSet.getString(1)) + "\t" + String.valueOf(resultSet.getString(2)));
  31.             }
  32.             statement.close();
  33.             connection.close();
  34.         } catch (SQLException e) {
  35.             e.printStackTrace();
  36.         }
  37.     }
  38. }

运行结果如下：

  1. connection success!
  2. Running:select * from test
  3. 1   lili    25
  4. 0   mary    15

SparkSQL对接SequoiaSQL

SparkSQL 可以通过 DataFrames 使用 JDBC 对 SequoiaSQL-MySQL 或 SequoiaSQL-PGSQL 进行读写操作。

对接前准备

  1. 下载相应的 JDBC 驱动，将其拷贝到 spark 集群 SPARK_HOME/jars 目录下

  2. 在读实例执行创建测试库、测试用户、授权及准备数据，在写实例执行创建测试库、测试用户及授权

    1. -- Create test database
    2. create database sparktest;
    4. -- Create a user representing your Spark cluster
    5. create user 'sparktest'@'%' identified by 'sparktest';
    7. -- Add privileges for the Spark cluster
    8. grant create, delete, drop, insert, select, update on sparktest.* to 'sparktest'@'%';
    9. flush privileges;
    11. -- Create a test table of physical characteristics.
    12. use sparktest;
    13. create table people (
    14.  id int(10) not null auto_increment,
    15.  name char(50) not null,
    16.  is_male tinyint(1) not null,
    17.  height_in int(4) not null,
    18.  weight_lb int(4) not null,
    19.  primary key (id),
    20.  key (id)
    21. );
    23. -- Create sample data to load into a DataFrame
    24. insert into people values (null, 'Alice', 0, 60, 125);
    25. insert into people values (null, 'Brian', 1, 64, 131);
    26. insert into people values (null, 'Charlie', 1, 74, 183);
    27. insert into people values (null, 'Doris', 0, 58, 102);
    28. insert into people values (null, 'Ellen', 0, 66, 140);
    29. insert into people values (null, 'Frank', 1, 66, 151);
    30. insert into people values (null, 'Gerard', 1, 68, 190);
    31. insert into people values (null, 'Harold', 1, 61, 128);

示例

  1. 编写示例代码

    1. package com.sequoiadb.test;
    3. import org.apache.spark.sql.Dataset;
    4. import org.apache.spark.sql.Row;
    5. import org.apache.spark.sql.SparkSession;
    7. import java.io.File;
    8. import java.io.FileInputStream;
    9. import java.util.Properties;
    11. public final class JDBCDemo {
    13. public static void main(String[] args) throws Exception {
    14.        String readUrl = "jdbc:mysql://192.168.30.81/sparktest" ;
    15.        String writeUrl = "jdbc:mysql://192.168.30.82/sparktest" ;
    17.     SparkSession spark = SparkSession.builder().appName("JDBCDemo").getOrCreate();
    19.     Properties dbProperties = new Properties();
    20.        dbProperties.setProperty("user", "sparktest") ;
    21.        dbProperties.setProperty("password", "sparktest" );
    23.     System.out.println("A DataFrame loaded from the entire contents of a table over JDBC.");
    24.     String where = "sparktest.people";
    25.     Dataset<Row> entireDF = spark.read().jdbc(readUrl, where, dbProperties);
    26.     entireDF.printSchema();
    27.     entireDF.show();
    29.     System.out.println("Filtering the table to just show the males.");
    30.     entireDF.filter("is_male = 1").show();
    32.     System.out.println("Alternately, pre-filter the table for males before loading over JDBC.");
    33.     where = "(select * from sparktest.people where is_male = 1) as subset";
    34.     Dataset<Row> malesDF = spark.read().jdbc(readUrl, where, dbProperties);
    35.     malesDF.show();
    37.     System.out.println("Update weights by 2 pounds (results in a new DataFrame with same column names)");
    38.     Dataset<Row> heavyDF = entireDF.withColumn("updated_weight_lb", entireDF.col("weight_lb").plus(2));
    39.     Dataset<Row> updatedDF = heavyDF.select("id", "name", "is_male", "height_in", "updated_weight_lb")
    40.         .withColumnRenamed("updated_weight_lb", "weight_lb");
    41.     updatedDF.show();
    43.     System.out.println("Save the updated data to a new table with JDBC");
    44.     where = "sparktest.updated_people";
    45.     updatedDF.write().mode("error").jdbc(writeUrl, where, dbProperties);
    47.     System.out.println("Load the new table into a new DataFrame to confirm that it was saved successfully.");
    48.     Dataset<Row> retrievedDF = spark.read().jdbc(writeUrl, where, dbProperties);
    49.     retrievedDF.show();
    51.     spark.stop();
    52.   }
    53. }

  2. 编译并提交任务

    1. mkdir -p target/java
    2. javac src/main/java/com/sequoiadb/test/JDBCDemo.java -classpath "$SPARK_HOME/jars/*" -d target/java
    3. cd target/java
    4. jar -cf ../JDBCDemo.jar *
    5. cd ../..
    6. APP_ARGS="--class com.sequoiadb.test.JDBCDemo target/JDBCDemo.jar"
    7. #本地提交
    8. $SPARK_HOME/bin/spark-submit --driver-class-path lib/mysql-connector-java-5.1.38.jar $APP_ARGS
    9. #集群提交
    10. $SPARK_HOME/bin/spark-submit --master spark://ip:7077 $APP_ARGS

    运行结果如下：

    1. A DataFrame loaded from the entire contents of a table over JDBC.
    2. root
    3. |-- id: integer (nullable = true)
    4. |-- name: string (nullable = true)
    5. |-- is_male: boolean (nullable = true)
    6. |-- height_in: integer (nullable = true)
    7. |-- weight_lb: integer (nullable = true)
    9. +---+-------+-------+---------+---------+
    10. | id|   name|is_male|height_in|weight_lb|
    11. +---+-------+-------+---------+---------+
    12. |  1|  Alice|  false|       60|      125|
    13. |  2|  Brian|   true|       64|      131|
    14. |  3|Charlie|   true|       74|      183|
    15. |  4|  Doris|  false|       58|      102|
    16. |  5|  Ellen|  false|       66|      140|
    17. |  6|  Frank|   true|       66|      151|
    18. |  7| Gerard|   true|       68|      190|
    19. |  8| Harold|   true|       61|      128|
    20. +---+-------+-------+---------+---------+
    22. Filtering the table to just show the males.
    23. +---+-------+-------+---------+---------+
    24. | id|   name|is_male|height_in|weight_lb|
    25. +---+-------+-------+---------+---------+
    26. |  2|  Brian|   true|       64|      131|
    27. |  3|Charlie|   true|       74|      183|
    28. |  6|  Frank|   true|       66|      151|
    29. |  7| Gerard|   true|       68|      190|
    30. |  8| Harold|   true|       61|      128|
    31. +---+-------+-------+---------+---------+
    33. Alternately, pre-filter the table for males before loading over JDBC.
    34. +---+-------+-------+---------+---------+
    35. | id|   name|is_male|height_in|weight_lb|
    36. +---+-------+-------+---------+---------+
    37. |  2|  Brian|   true|       64|      131|
    38. |  3|Charlie|   true|       74|      183|
    39. |  6|  Frank|   true|       66|      151|
    40. |  7| Gerard|   true|       68|      190|
    41. |  8| Harold|   true|       61|      128|
    42. +---+-------+-------+---------+---------+
    44. Update weights by 2 pounds (results in a new DataFrame with same column names)
    45. +---+-------+-------+---------+---------+
    46. | id|   name|is_male|height_in|weight_lb|
    47. +---+-------+-------+---------+---------+
    48. |  1|  Alice|  false|       60|      127|
    49. |  2|  Brian|   true|       64|      133|
    50. |  3|Charlie|   true|       74|      185|
    51. |  4|  Doris|  false|       58|      104|
    52. |  5|  Ellen|  false|       66|      142|
    53. |  6|  Frank|   true|       66|      153|
    54. |  7| Gerard|   true|       68|      192|
    55. |  8| Harold|   true|       61|      130|
    56. +---+-------+-------+---------+---------+
    58. Save the updated data to a new table with JDBC
    59. Load the new table into a new DataFrame to confirm that it was saved successfully.
    60. +---+-------+-------+---------+---------+
    61. | id|   name|is_male|height_in|weight_lb|
    62. +---+-------+-------+---------+---------+
    63. |  1|  Alice|  false|       60|      127|
    64. |  2|  Brian|   true|       64|      133|
    65. |  3|Charlie|   true|       74|      185|
    66. |  4|  Doris|  false|       58|      104|
    67. |  5|  Ellen|  false|       66|      142|
    68. |  6|  Frank|   true|       66|      153|
    69. |  7| Gerard|   true|       68|      192|
    70. |  8| Harold|   true|       61|      130|
    71. +---+-------+-------+---------+---------+