[
https://issues.apache.org/jira/browse/SPARK-22481?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16248837#comment-16248837
]
Ran Haim commented on SPARK-22481:
----------------------------------
Hi,
I create this simple test that will show how slow refresh table can be.
I create a parquet table, with 10K files just to make it obvious how slow it
can get - but a table with a lot of partitions will have the same effect.
{code:java}
import java.util.UUID
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import org.scalatest.{BeforeAndAfter, FunSuite}
@RunWith(classOf[JUnitRunner])
class TestRefreshTable extends FunSuite with BeforeAndAfter{
{
if (System.getProperty("os.name").toLowerCase.startsWith("windows")) {
val hadoopDirURI = ClassLoader.getSystemResource("hadoop").toURI.getPath
System.setProperty("hadoop.home.dir",hadoopDirURI )
}
}
var fs: FileSystem = _
before{
fs = FileSystem.get(new Configuration())
fs.setWriteChecksum(false)
}
private val path = "/tmp/test"
test("big table refresh test"){
val config = new SparkConf().
setMaster("local[*]").
setAppName("test").
set("spark.ui.enabled", "false")
val sparkSession = SparkSession
.builder()
.config(config)
.enableHiveSupport()
.getOrCreate()
try {
for (i <- 1 to 10000) {
val id = UUID.randomUUID().toString
fs.create(new Path(path, id))
}
sparkSession.sql("drop TABLE if exists test")
sparkSession.sql(
s"""
CREATE EXTERNAL TABLE test(
`rowId` bigint)
ROW FORMAT SERDE
'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
'$path'
""")
val start = System.currentTimeMillis()
sparkSession.catalog.refreshTable("test")
val end = System.currentTimeMillis()
println(s"refresh took ${end - start}")
}finally {
sparkSession.close()
}
}
}
{code}
> CatalogImpl.refreshTable is slow
> --------------------------------
>
> Key: SPARK-22481
> URL: https://issues.apache.org/jira/browse/SPARK-22481
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 2.1.1, 2.1.2, 2.2.0
> Reporter: Ran Haim
> Priority: Critical
>
> CatalogImpl.refreshTable was updated in 2.1.1 and since than it has become
> really slow.
> The cause of the issue is that it is now *always* creates a dataset, and this
> is redundant most of the time, we only need the dataset if the table is
> cached.
> before 2.1.1:
> override def refreshTable(tableName: String): Unit = {
> val tableIdent =
> sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
> // Temp tables: refresh (or invalidate) any metadata/data cached in the
> plan recursively.
> // Non-temp tables: refresh the metadata cache.
> sessionCatalog.refreshTable(tableIdent)
> // If this table is cached as an InMemoryRelation, drop the original
> // cached version and make the new version cached lazily.
> val logicalPlan =
> sparkSession.sessionState.catalog.lookupRelation(tableIdent)
> // Use lookupCachedData directly since RefreshTable also takes
> databaseName.
> val isCached =
> sparkSession.sharedState.cacheManager.lookupCachedData(logicalPlan).nonEmpty
> if (isCached) {
> // Create a data frame to represent the table.
> // TODO: Use uncacheTable once it supports database name.
> {color:red} val df = Dataset.ofRows(sparkSession, logicalPlan){color}
> // Uncache the logicalPlan.
> sparkSession.sharedState.cacheManager.uncacheQuery(df, blocking = true)
> // Cache it again.
> sparkSession.sharedState.cacheManager.cacheQuery(df,
> Some(tableIdent.table))
> }
> }
> after 2.1.1:
> override def refreshTable(tableName: String): Unit = {
> val tableIdent =
> sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
> // Temp tables: refresh (or invalidate) any metadata/data cached in the
> plan recursively.
> // Non-temp tables: refresh the metadata cache.
> sessionCatalog.refreshTable(tableIdent)
> // If this table is cached as an InMemoryRelation, drop the original
> // cached version and make the new version cached lazily.
> {color:red} val table = sparkSession.table(tableIdent){color}
> if (isCached(table)) {
> // Uncache the logicalPlan.
> sparkSession.sharedState.cacheManager.uncacheQuery(table, blocking =
> true)
> // Cache it again.
> sparkSession.sharedState.cacheManager.cacheQuery(table,
> Some(tableIdent.table))
> }
> }
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
