You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2023/10/10 08:17:20 UTC
[spark] branch branch-3.5 updated: [SPARK-45475][SQL] Uses DataFrame.foreachPartition instead of RDD.foreachPartition in JdbcUtils
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 40d44a3fa4a [SPARK-45475][SQL] Uses DataFrame.foreachPartition instead of RDD.foreachPartition in JdbcUtils
40d44a3fa4a is described below
commit 40d44a3fa4a59f6e98d076a3b3f006851f25be3a
Author: Hyukjin Kwon <gu...@apache.org>
AuthorDate: Tue Oct 10 17:16:11 2023 +0900
[SPARK-45475][SQL] Uses DataFrame.foreachPartition instead of RDD.foreachPartition in JdbcUtils
This PR is kind of a followup for https://github.com/apache/spark/pull/39976 that addresses https://github.com/apache/spark/pull/39976#issuecomment-1752930380 comment.
In order to probably assign the SQL execution ID so `df.observe` works with this.
Yes. `df.observe` will work with JDBC connectors.
Manually tested.
Unit test was added.
Closes #43304 from HyukjinKwon/foreachbatch.
Authored-by: Hyukjin Kwon <gu...@apache.org>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
(cherry picked from commit 39cc4abaff73cb49f9d79d1d844fe5c9fa14c917)
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../sql/execution/datasources/jdbc/JdbcUtils.scala | 2 +-
.../scala/org/apache/spark/sql/jdbc/JDBCSuite.scala | 17 ++++++++++++++++-
2 files changed, 17 insertions(+), 2 deletions(-)
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 b7019c1dcbe..aae4be3f094 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
@@ -898,7 +898,7 @@ object JdbcUtils extends Logging with SQLConfHelper {
case Some(n) if n < df.rdd.getNumPartitions => df.coalesce(n)
case _ => df
}
- repartitionedDF.rdd.foreachPartition { iterator => savePartition(
+ repartitionedDF.foreachPartition { iterator => savePartition(
table, iterator, rddSchema, insertStmt, batchSize, dialect, isolationLevel, options)
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index 93b6652d516..eae171e20b7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
@@ -29,7 +29,7 @@ import org.mockito.ArgumentMatchers._
import org.mockito.Mockito._
import org.apache.spark.{SparkException, SparkSQLException}
-import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Observation, QueryTest, Row}
import org.apache.spark.sql.catalyst.{analysis, TableIdentifier}
import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.ShowCreateTable
@@ -39,6 +39,7 @@ import org.apache.spark.sql.execution.command.{ExplainCommand, ShowCreateTableCo
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JDBCPartition, JDBCRelation, JdbcUtils}
import org.apache.spark.sql.execution.metric.InputOutputMetricsHelper
+import org.apache.spark.sql.functions.{lit, percentile_approx}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.test.SharedSparkSession
@@ -2057,4 +2058,18 @@ class JDBCSuite extends QueryTest with SharedSparkSession {
val df = sql("SELECT * FROM composite_name WHERE `last name` = 'smith'")
assert(df.collect.toSet === Set(Row("smith", 1)))
}
+
+ test("SPARK-45475: saving a table via JDBC should work with observe API") {
+ val tableName = "test_table"
+ val namedObservation = Observation("named")
+ val observed_df = spark.range(100).observe(
+ namedObservation, percentile_approx($"id", lit(0.5), lit(100)).as("percentile_approx_val"))
+
+ observed_df.write.format("jdbc")
+ .option("url", urlWithUserAndPass)
+ .option("dbtable", tableName).save()
+
+ val expected = Map("percentile_approx_val" -> 49)
+ assert(namedObservation.get === expected)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org