官网链接：
http://spark.apache.org/docs/latest/sql-programming-guide.html#jdbc-to-other-databases
http://spark.apache.org/docs/latest/sql-data-sources-jdbc.html
1. 过滤数据
情景：使用spark通过JDBC的方式读取postgresql数据库中的表然后存储到hive表*后面数据处理使用，但是只读取postgresql表中的某些字段，并且做一下数据上的过滤
根据平常的方式，基本都是读取整张表，感觉不应该这么不友好的，于是去官网翻了翻，如下：
指定dbtable参数时候可以使用子查询的方式，不单纯是指定表名
测试代码如下：
package com.kong.test.test;
import java.util.Properties;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
public class SparkHiveTest {
public static void main(String[] args) {
SparkSession spark = SparkSession
.builder()
.appName("SparkCalibration")
.master("local")
.enableHiveSupport()
.getOrCreate();
spark.sparkContext().setLogLevel("ERROR");
spark.sparkContext().setLocalProperty("spark.scheduler.pool", "production");
String t2 = "(select id, name from test1) tmp";//这里需要有个别名
String createSql = "create table if not exists default.test1 (\r\n" +
"id string,\r\n" +
"name string\r\n" +
")ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' stored as TEXTFILE";
spark.sql(createSql);
spark.read().format("jdbc")
.option("url", "jdbc:postgresql://ip address/database")
.option("dbtable", t2).option("user", "login user").option("password", "login passwd")
.option("fetchsize", "1000")
.load()
.createOrReplaceTempView("test1_tmp");
spark.sql("insert overwrite table default.test1 select * from test1_tmp").show();
}
}
另外：如果对于hive表的存储格式没有要求，可以更简洁，如下：
spark.read().format("jdbc")
.option("url", "jdbc:postgresql://ip address/database")
.option("dbtable", t2).option("user", "login user").option("password", "login passwd")
.option("fetchsize", "1000")
.load().write().mode(SaveMode.Overwrite).saveAsTable("default.test");
至于基于哪种保存模式（SaveMode.Overwrite）可以结合实际场景；另外spark saveAsTable（）默认是以parquet+snappy的形式写数据（生成的文件名.snappy.parquet），当然，也可以通过format（）传入参数，使用orc等格式，并且可以指定其他压缩方式。
2. spark通过JDBC读取外部数据库的源码实现
2.1 最简洁的api，单分区
源码如下：
/**
* Construct a `DataFrame` representing the database table accessible via JDBC URL
* url named table and connection properties.
*
* @since 1.4.0
*/
def jdbc(url: String, table: String, properties: Properties): DataFrame = {
assertNoSpecifiedSchema("jdbc")
// properties should override settings in extraOptions.
this.extraOptions ++= properties.asScala
// explicit url and dbtable should override all
this.extraOptions += (JDBCOptions.JDBC_URL -> url, JDBCOptions.JDBC_TABLE_NAME -> table)
format("jdbc").load()
}
2.2 指定表某个字段的上下限值（数值类型），生成相对应的where条件并行读取，源码如下：
/**
* Construct a `DataFrame` representing the database table accessible via JDBC URL
* url named table. Partitions of the table will be retrieved in parallel based on the parameters
* passed to this function.
*
* Don't create too many partitions in parallel on a large cluster; otherwise Spark might crash
* your external database systems.
*
* @param url JDBC database url of the form `jdbc:subprotocol:subname`.
* @param table Name of the table in the external database.
* @param columnName the name of a column of integral type that will be used for partitioning.
* @param lowerBound the minimum value of `columnName` used to decide partition stride.
* @param upperBound the maximum value of `columnName` used to decide partition stride.
* @param numPartitions the number of partitions. This, along with `lowerBound` (inclusive),
* `upperBound` (exclusive), form partition strides for generated WHERE
* clause expressions used to split the column `columnName` evenly. When
* the input is less than 1, the number is set to 1.
* @param connectionProperties JDBC database connection arguments, a list of arbitrary string
* tag/value. Normally at least a "user" and "password" property
* should be included. "fetchsize" can be used to control the
* number of rows per fetch.
* @since 1.4.0
*/
def jdbc(
url: String,
table: String,
columnName: String,
lowerBound: Long,
upperBound: Long,
numPartitions: Int,
connectionProperties: Properties): DataFrame = {
// columnName, lowerBound, upperBound and numPartitions override settings in extraOptions.
this.extraOptions ++= Map(
JDBCOptions.JDBC_PARTITION_COLUMN -> columnName,
JDBCOptions.JDBC_LOWER_BOUND -> lowerBound.toString,
JDBCOptions.JDBC_UPPER_BOUND -> upperBound.toString,
JDBCOptions.JDBC_NUM_PARTITIONS -> numPartitions.toString)
jdbc(url, table, connectionProperties)
}
2.3 通过predicates: Array[String],传入每个分区的where子句中的谓词条件，并行读取，比如 ：
String[] predicates = new String[] {"date <= '20180501'","date > '20180501' and date <= '20181001'","date > '20181001'"};
/**
* Construct a `DataFrame` representing the database table accessible via JDBC URL
* url named table using connection properties. The `predicates` parameter gives a list
* expressions suitable for inclusion in WHERE clauses; each one defines one partition
* of the `DataFrame`.
*
* Don't create too many partitions in parallel on a large cluster; otherwise Spark might crash
* your external database systems.
*
* @param url JDBC database url of the form `jdbc:subprotocol:subname`
* @param table Name of the table in the external database.
* @param predicates Condition in the where clause for each partition.
* @param connectionProperties JDBC database connection arguments, a list of arbitrary string
* tag/value. Normally at least a "user" and "password" property
* should be included. "fetchsize" can be used to control the
* number of rows per fetch.
* @since 1.4.0
*/
def jdbc(
url: String,
table: String,
predicates: Array[String],
connectionProperties: Properties): DataFrame = {
assertNoSpecifiedSchema("jdbc")
// connectionProperties should override settings in extraOptions.
val params = extraOptions.toMap ++ connectionProperties.asScala.toMap
val options = new JDBCOptions(url, table, params)
val parts: Array[Partition] = predicates.zipWithIndex.map { case (part, i) =>
JDBCPartition(part, i) : Partition
}
val relation = JDBCRelation(parts, options)(sparkSession)
sparkSession.baseRelationToDataFrame(relation)
}