Skip to content

[SPARK-18413][SQL] Add maxConnections JDBCOption #15868

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dongjoon-hyun wants to merge 7 commits into from
Closed

[SPARK-18413][SQL] Add maxConnections JDBCOption #15868

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
57b10ec
[SPARK-18413][SQL] Control the number of JDBC connections by repartit…
dongjoon-hyun Nov 13, 2016
bf1d6bd
Add a negative test case.
dongjoon-hyun Nov 17, 2016
28718d0
Update doc.
dongjoon-hyun Nov 18, 2016
0391dbf
Update docs
dongjoon-hyun Nov 19, 2016
ca75a4e
Update docs.
dongjoon-hyun Nov 19, 2016
cbb74d9
Use Option[Int].
dongjoon-hyun Nov 21, 2016
dc152f4
Use Option[Int] really.
dongjoon-hyun Nov 21, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions docs/sql-programming-guide.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1086,6 +1086,13 @@ the following case-sensitive options:
</td>
</tr>

<tr>
<td><code>maxConnections</code></td>
<td>
The number of JDBC connections, which specifies the maximum number of simultaneous JDBC connections that are allowed. This option applies only to writing. It defaults to the number of partitions of RDD.
</td>
</tr>

<tr>
<td><code>isolationLevel</code></td>
<td>
Expand Down
6 changes: 6 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,11 @@ class JDBCOptions(
case "REPEATABLE_READ" => Connection.TRANSACTION_REPEATABLE_READ
case "SERIALIZABLE" => Connection.TRANSACTION_SERIALIZABLE
}
// the maximum number of connections
val maxConnections = parameters.getOrElse(JDBC_MAX_CONNECTIONS, null)
Copy link
Contributor

@cloud-fan cloud-fan Nov 21, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we should avoid using null in scala, how about make maxConnections Option[Int]?

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 21, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure! No problem. I'll update this PR again.

require(maxConnections == null || maxConnections.toInt > 0,
s"Invalid value `$maxConnections` for parameter `$JDBC_MAX_CONNECTIONS`. " +
s"The minimum value is 1.")
Copy link
Member

@gatorsmile gatorsmile Nov 17, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: no need to add s

}

object JDBCOptions {
Expand All @@ -144,4 +149,5 @@ object JDBCOptions {
val JDBC_CREATE_TABLE_OPTIONS = newOption("createTableOptions")
val JDBC_BATCH_INSERT_SIZE = newOption("batchsize")
val JDBC_TXN_ISOLATION_LEVEL = newOption("isolationLevel")
val JDBC_MAX_CONNECTIONS = newOption("maxConnections")
}
9 changes: 8 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -667,7 +667,14 @@ object JdbcUtils extends Logging {
val getConnection: () => Connection = createConnectionFactory(options)
val batchSize = options.batchSize
val isolationLevel = options.isolationLevel
df.foreachPartition(iterator => savePartition(
val maxConnections = options.maxConnections
val repartitionedDF =
if (maxConnections != null && maxConnections.toInt < df.rdd.getNumPartitions) {
Copy link
Member

@gatorsmile gatorsmile Nov 16, 2016 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the input is less than 1, we should detect it at the beginning, right?

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 16, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yep. I'll add require like batchsize.

Copy link
Member Author

@dongjoon-hyun dongjoon-hyun Nov 16, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's done.

df.coalesce(maxConnections.toInt)
} else {
df
}
repartitionedDF.foreachPartition(iterator => savePartition(
getConnection, table, iterator, rddSchema, nullTypes, batchSize, dialect, isolationLevel)
)
}
Expand Down