[SPARK-47585][SQL] SQL core: Migrate logInfo with variables to structured logging framework #46264

panbingkun · 2024-04-28T11:45:00Z

What changes were proposed in this pull request?

The pr aims to migrate logInfo in module SQL core with variables to structured logging framework.

Why are the changes needed?

To enhance Apache Spark's logging system by implementing structured logging.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Pass GA.

Was this patch authored or co-authored using generative AI tooling?

No.

…ured logging framework

panbingkun · 2024-04-29T01:26:14Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FilePartitionReader.scala

@@ -39,7 +39,7 @@ class FilePartitionReader[T](
    if (currentReader == null) {
      if (files.hasNext) {
        val file = files.next()
-        logInfo(s"Reading file $file")
+        logInfo(log"Reading file ${MDC(PARTITIONED_FILE, file)}")


The class of file is PartitionedFile, So LogKeys.PATH is not used

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala

Lines 70 to 72 in 8c446f3

override def toString: String = {

s"path: $filePath, range: $start-${start + length}, partition values: $partitionValues"

}

CURRENT_FILE

panbingkun · 2024-04-29T01:30:37Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileScanRDD.scala

@@ -233,7 +233,7 @@ class FileScanRDD(
        if (files.hasNext) {
          currentFile = files.next()
          updateMetadataRow()
-          logInfo(s"Reading File $currentFile")
+          logInfo(log"Reading File ${MDC(PARTITIONED_FILE, currentFile)}")


The class of currentFile is PartitionedFile.

CURRENT_FILE seems better

panbingkun · 2024-04-29T01:33:19Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala

@@ -494,7 +495,7 @@ object DataSourceStrategy
      val partitionSet = AttributeSet(partitionColumns)
      val predicates = ExpressionSet(normalizedFilters
        .flatMap(extractPredicatesWithinOutputSet(_, partitionSet)))
-      logInfo(s"Pruning directories with: ${predicates.mkString(",")}")
+      logInfo(log"Pruning directories with: ${MDC(PREDICATES, predicates.mkString(","))}")


I'm not sure if EXPRS is more general.

PREDICATES looks fine

panbingkun · 2024-04-29T01:44:57Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -468,7 +467,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
        } else if (filteredRules.nonEmpty) {
          Some(Batch(batch.name, batch.strategy, filteredRules: _*))
        } else {
-          logInfo(log"Optimization batch '${MDC(RULE_BATCH_NAME, batch.name)}' " +
+          logInfo(log"Optimization batch '${MDC(LogKeys.RULE_BATCH_NAME, batch.name)}' " +


Unified as LogKeys.BATCH_NAME, remove LogKeys.RULE_BATCH_NAME

panbingkun · 2024-04-29T01:51:00Z

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggregateCodegenSupport.scala

        if (Utils.isTesting) {
-          throw SparkException.internalError(errMsg)
+          throw SparkException.internalError(errMsg.message)
        } else {
          logInfo(errMsg)


panbingkun · 2024-04-29T02:47:35Z

sql/core/src/main/scala/org/apache/spark/sql/execution/r/ArrowRRunner.scala

-                  compute,
-                  output,
-                  boot + init + broadcast + input + compute + output))
+              logInfo(log"Times: boot = ${MDC(LogKeys.BOOT, format(boot))} s, " +


I'm not sure whether to prefix the following keys as R_

panbingkun · 2024-04-29T03:03:55Z

common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala

  case object RMSE extends LogKey
  case object ROCKS_DB_LOG_LEVEL extends LogKey
  case object ROCKS_DB_LOG_MESSAGE extends LogKey
  case object RPC_ENDPOINT_REF extends LogKey
-  case object RULE_BATCH_NAME extends LogKey


Unified as LogKeys.BATCH_NAME, remove LogKeys.RULE_BATCH_NAME

panbingkun · 2024-04-29T03:05:00Z

common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala

  case object PARTITIONED_FILE_READER extends LogKey
-  case object PARTITIONS_SIZE extends LogKey


Unified as LogKeys.COUNT, remove LogKeys.PARTITIONS_SIZE

PARTITIONS_SIZE seems useful. Shall we keep it?

PARTITIONS_SIZE -> NUM_PARTITIONS

It has been changed to NUM_PARTITIONS as suggested below
At the same time, I unified LogKeys.NUM_PARTITION to LogKeys.NUM_PARTITIONS, and remove LogKeys.NUM_PARTITION

panbingkun · 2024-04-29T03:05:24Z

common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala

@@ -191,20 +203,23 @@ object LogKeys {
  case object HIVE_OPERATION_TYPE extends LogKey
  case object HOST extends LogKey
  case object HOST_PORT extends LogKey
-  case object IDENTIFIER extends LogKey


Unified as LogKeys.TABLE_NAME, remove LogKeys.IDENTIFIER

Currently, only two places use this key, and they all mean the name of the table.

Thanks for unifying them!

panbingkun · 2024-04-29T03:06:02Z

common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala

@@ -111,7 +117,7 @@ object LogKeys {
  case object DATA_FILE_NUM extends LogKey
  case object DATA_SOURCE extends LogKey
  case object DATA_SOURCES extends LogKey
-  case object DATA_SOURCE_PROVIDER extends LogKey


Unified as LogKeys.DATA_SOURCE, remove LogKeys.DATA_SOURCE_PROVIDER

panbingkun · 2024-04-29T03:12:51Z

cc @gengliangwang

gengliangwang · 2024-04-29T19:58:43Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala

@@ -717,14 +716,15 @@ case class RepairTableCommand(
          evalPool.shutdown()
        }
      val total = partitionSpecsAndLocs.length
-      logInfo(s"Found $total partitions in $root")
+      logInfo(log"Found ${MDC(LogKeys.TOTAL, total)} partitions in ${MDC(LogKeys.PATH, root)}")


NUM_PARTITIONS

gengliangwang · 2024-04-29T19:58:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/ddl.scala


      val partitionStats = if (spark.sessionState.conf.gatherFastStats) {
        gatherPartitionStats(spark, partitionSpecsAndLocs, fs, pathFilter, threshold)
      } else {
        Map.empty[Path, PartitionStatistics]
      }
-      logInfo(s"Finished to gather the fast stats for all $total partitions.")
+      logInfo(log"Finished to gather the fast stats for all ${MDC(LogKeys.TOTAL, total)} " +


gengliangwang · 2024-04-29T20:06:45Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala

@@ -99,7 +99,7 @@ object FilePartition extends Logging {
      val desiredSplitBytes =
        (totalSizeInBytes / BigDecimal(maxPartNum.get)).setScale(0, RoundingMode.UP).longValue
      val desiredPartitions = getFilePartitions(partitionedFiles, desiredSplitBytes, openCostBytes)
-      logWarning(log"The number of partitions is ${MDC(PARTITIONS_SIZE, partitions.size)}, " +
+      logWarning(log"The number of partitions is ${MDC(COUNT, partitions.size)}, " +


gengliangwang · 2024-04-29T20:07:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala

-    logInfo {
-      s"Pruned ${numBuckets - numBucketsSelected} out of $numBuckets buckets."
-    }
+    logInfo(log"Pruned ${MDC(COUNT, numBuckets - numBucketsSelected)} " +


gengliangwang · 2024-04-29T20:11:32Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRelation.scala

@@ -164,8 +164,10 @@ private[sql] object JDBCRelation extends Logging {
      i = i + 1
    }
    val partitions = ans.toArray
-    logInfo(s"Number of partitions: $numPartitions, WHERE clauses of these partitions: " +
-      partitions.map(_.asInstanceOf[JDBCPartition].whereClause).mkString(", "))
+    logInfo(log"Number of partitions: ${MDC(COUNT, numPartitions)}, WHERE clauses of " +


gengliangwang · 2024-04-29T20:18:30Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

@@ -421,7 +424,8 @@ case class EnsureRequirements(
            .mergePartitions(leftSpec.partitioning, rightSpec.partitioning, partitionExprs)
            .map(v => (v, 1))

-        logInfo(s"After merging, there are ${mergedPartValues.size} partitions")
+        logInfo(
+          log"After merging, there are ${MDC(LogKeys.COUNT, mergedPartValues.size)} partitions")


gengliangwang · 2024-04-29T20:26:59Z

LGTM except for a few minor comments on the names.

panbingkun · 2024-04-30T01:28:51Z

LGTM except for a few minor comments on the names.

All done. Thanks!

gengliangwang

Thanks for the works!

…ured logging framework ### What changes were proposed in this pull request? The pr aims to migrate `logInfo` in module `SQL core` with variables to `structured logging framework`. ### Why are the changes needed? To enhance Apache Spark's logging system by implementing structured logging. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? - Pass GA. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#46264 from panbingkun/SPARK-47585. Authored-by: panbingkun <panbingkun@baidu.com> Signed-off-by: Gengliang Wang <gengliang@apache.org>

panbingkun added 2 commits April 28, 2024 15:19

test

d800641

[SPARK-47585][SQL] SQL core: Migrate logInfo with variables to struct…

757f721

…ured logging framework

github-actions bot added SQL STRUCTURED STREAMING PYTHON R labels Apr 28, 2024

panbingkun marked this pull request as draft April 28, 2024 11:46

fix

de4aa03

panbingkun commented Apr 29, 2024

View reviewed changes

panbingkun added 3 commits April 29, 2024 10:55

update

f658578

Merge branch 'master' into SPARK-47585

8ad088b

update

0fd3746

panbingkun commented Apr 29, 2024

View reviewed changes

update

30d7f95

panbingkun marked this pull request as ready for review April 29, 2024 03:12

gengliangwang reviewed Apr 29, 2024

View reviewed changes

panbingkun added 2 commits April 30, 2024 08:08

Merge branch 'master' into SPARK-47585

5540de3

fix

764c660

github-actions bot added the ML label Apr 30, 2024

gengliangwang approved these changes Apr 30, 2024

View reviewed changes

gengliangwang closed this in 87b20b1 Apr 30, 2024

	override def toString: String = {
	s"path: $filePath, range: $start-${start + length}, partition values: $partitionValues"
	}

		case object PARTITIONED_FILE_READER extends LogKey
		case object PARTITIONS_SIZE extends LogKey

[SPARK-47585][SQL] SQL core: Migrate logInfo with variables to structured logging framework #46264

[SPARK-47585][SQL] SQL core: Migrate logInfo with variables to structured logging framework #46264

Conversation

panbingkun commented Apr 28, 2024 • edited

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

panbingkun commented Apr 29, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gengliangwang commented Apr 29, 2024

panbingkun commented Apr 30, 2024

gengliangwang left a comment

Choose a reason for hiding this comment

panbingkun commented Apr 28, 2024 •

edited