You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2016/07/21 08:23:45 UTC
spark git commit: [SPARK-16226][SQL] Weaken JDBC isolation level to
avoid locking when writing partitions
Repository: spark
Updated Branches:
refs/heads/master 1bf13ba3a -> 864b764ea
[SPARK-16226][SQL] Weaken JDBC isolation level to avoid locking when writing partitions
## What changes were proposed in this pull request?
Saving partitions to JDBC in transaction can use a weaker transaction isolation level to reduce locking. Use better method to check if transactions are supported.
## How was this patch tested?
Existing Jenkins tests.
Author: Sean Owen <so...@cloudera.com>
Closes #14054 from srowen/SPARK-16226.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/864b764e
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/864b764e
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/864b764e
Branch: refs/heads/master
Commit: 864b764eafa57a1418b683ccf6899b01bab28fba
Parents: 1bf13ba
Author: Sean Owen <so...@cloudera.com>
Authored: Thu Jul 21 09:23:41 2016 +0100
Committer: Sean Owen <so...@cloudera.com>
Committed: Thu Jul 21 09:23:41 2016 +0100
----------------------------------------------------------------------
.../org/apache/spark/sql/DataFrameWriter.scala | 6 ++-
.../execution/datasources/jdbc/JdbcUtils.scala | 54 +++++++++++++++-----
2 files changed, 46 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/864b764e/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index 12b3046..e6a8dfa 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -392,7 +392,11 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
* @param connectionProperties JDBC database connection arguments, a list of arbitrary string
* tag/value. Normally at least a "user" and "password" property
* should be included. "batchsize" can be used to control the
- * number of rows per insert.
+ * number of rows per insert. "isolationLevel" can be one of
+ * "NONE", "READ_COMMITTED", "READ_UNCOMMITTED", "REPEATABLE_READ",
+ * or "SERIALIZABLE", corresponding to standard transaction
+ * isolation levels defined by JDBC's Connection object, with default
+ * of "READ_UNCOMMITTED".
* @since 1.4.0
*/
def jdbc(url: String, table: String, connectionProperties: Properties): Unit = {
http://git-wip-us.apache.org/repos/asf/spark/blob/864b764e/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
index d3e1efc..ce71a7d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
@@ -37,6 +37,7 @@ object JdbcUtils extends Logging {
// the property names are case sensitive
val JDBC_BATCH_FETCH_SIZE = "fetchsize"
val JDBC_BATCH_INSERT_SIZE = "batchsize"
+ val JDBC_TXN_ISOLATION_LEVEL = "isolationLevel"
/**
* Returns a factory for creating connections to the given JDBC URL.
@@ -139,8 +140,8 @@ object JdbcUtils extends Logging {
/**
* Saves a partition of a DataFrame to the JDBC database. This is done in
- * a single database transaction in order to avoid repeatedly inserting
- * data as much as possible.
+ * a single database transaction (unless isolation level is "NONE")
+ * in order to avoid repeatedly inserting data as much as possible.
*
* It is still theoretically possible for rows in a DataFrame to be
* inserted into the database more than once if a stage somehow fails after
@@ -158,25 +159,44 @@ object JdbcUtils extends Logging {
rddSchema: StructType,
nullTypes: Array[Int],
batchSize: Int,
- dialect: JdbcDialect): Iterator[Byte] = {
+ dialect: JdbcDialect,
+ isolationLevel: Int): Iterator[Byte] = {
require(batchSize >= 1,
s"Invalid value `${batchSize.toString}` for parameter " +
s"`${JdbcUtils.JDBC_BATCH_INSERT_SIZE}`. The minimum value is 1.")
val conn = getConnection()
var committed = false
- val supportsTransactions = try {
- conn.getMetaData().supportsDataManipulationTransactionsOnly() ||
- conn.getMetaData().supportsDataDefinitionAndDataManipulationTransactions()
- } catch {
- case NonFatal(e) =>
- logWarning("Exception while detecting transaction support", e)
- true
+
+ var finalIsolationLevel = Connection.TRANSACTION_NONE
+ if (isolationLevel != Connection.TRANSACTION_NONE) {
+ try {
+ val metadata = conn.getMetaData
+ if (metadata.supportsTransactions()) {
+ // Update to at least use the default isolation, if any transaction level
+ // has been chosen and transactions are supported
+ val defaultIsolation = metadata.getDefaultTransactionIsolation
+ finalIsolationLevel = defaultIsolation
+ if (metadata.supportsTransactionIsolationLevel(isolationLevel)) {
+ // Finally update to actually requested level if possible
+ finalIsolationLevel = isolationLevel
+ } else {
+ logWarning(s"Requested isolation level $isolationLevel is not supported; " +
+ s"falling back to default isolation level $defaultIsolation")
+ }
+ } else {
+ logWarning(s"Requested isolation level $isolationLevel, but transactions are unsupported")
+ }
+ } catch {
+ case NonFatal(e) => logWarning("Exception while detecting transaction support", e)
+ }
}
+ val supportsTransactions = finalIsolationLevel != Connection.TRANSACTION_NONE
try {
if (supportsTransactions) {
conn.setAutoCommit(false) // Everything in the same db transaction.
+ conn.setTransactionIsolation(finalIsolationLevel)
}
val stmt = insertStatement(conn, table, rddSchema, dialect)
try {
@@ -285,9 +305,17 @@ object JdbcUtils extends Logging {
val rddSchema = df.schema
val getConnection: () => Connection = createConnectionFactory(url, properties)
val batchSize = properties.getProperty(JDBC_BATCH_INSERT_SIZE, "1000").toInt
- df.foreachPartition { iterator =>
- savePartition(getConnection, table, iterator, rddSchema, nullTypes, batchSize, dialect)
- }
+ val isolationLevel =
+ properties.getProperty(JDBC_TXN_ISOLATION_LEVEL, "READ_UNCOMMITTED") match {
+ case "NONE" => Connection.TRANSACTION_NONE
+ case "READ_UNCOMMITTED" => Connection.TRANSACTION_READ_UNCOMMITTED
+ case "READ_COMMITTED" => Connection.TRANSACTION_READ_COMMITTED
+ case "REPEATABLE_READ" => Connection.TRANSACTION_REPEATABLE_READ
+ case "SERIALIZABLE" => Connection.TRANSACTION_SERIALIZABLE
+ }
+ df.foreachPartition(iterator => savePartition(
+ getConnection, table, iterator, rddSchema, nullTypes, batchSize, dialect, isolationLevel)
+ )
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org