forked from Azure/azure-sdk-for-java
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fixing maxItemCount from Spark to control batch Size of query requests (
Azure#23963) * Fixing maxItemCount from Spark to control batch Size of query requests * Fixing build warnings * Adding more robust temp file cleanup * Fixing test issues * Reducing BulkWriter.DefaultMaxPendingOperationPerCore
- Loading branch information
1 parent
7f0dd99
commit eb3b7d2
Showing
8 changed files
with
266 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
57 changes: 57 additions & 0 deletions
57
...-spark_3-1_2-12/src/test/scala/com/azure/cosmos/spark/SimpleFileDiagnosticsProvider.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
| package com.azure.cosmos.spark | ||
|
|
||
| import com.azure.cosmos.spark.diagnostics.{DiagnosticsProvider, ILogger} | ||
| import org.apache.commons.io.FileUtils | ||
|
|
||
| import java.io.{File, IOException} | ||
| import java.nio.file.Files | ||
| import scala.collection.concurrent.TrieMap | ||
|
|
||
| // only when diagnostics enabled, | ||
| // - logs each individual writes success and/or failures with id,pk | ||
| // - logs each documentServiceRequest and documentServiceResponse | ||
| private[spark] class SimpleFileDiagnosticsProvider extends DiagnosticsProvider { | ||
| override def getLogger(classType: Class[_]): ILogger = | ||
| SimpleFileDiagnosticsProvider.getOrCreateSingletonLoggerInstance(classType) | ||
| } | ||
|
|
||
| private[spark] object SimpleFileDiagnosticsProvider { | ||
| val singletonInstances: TrieMap[Class[_], SimpleFileDiagnosticsSlf4jLogger] = | ||
| new TrieMap[Class[_], SimpleFileDiagnosticsSlf4jLogger]() | ||
|
|
||
| private val folderName = System.getProperty("java.io.tmpdir") + "/SimpleFileDiagnostics" | ||
| private val folder = new File(folderName) | ||
| FileUtils.forceDeleteOnExit(folder) | ||
|
|
||
| def getOrCreateSingletonLoggerInstance(classType: Class[_]): SimpleFileDiagnosticsSlf4jLogger = { | ||
| if (!folder.exists()) { | ||
| folder.mkdirs() | ||
| } | ||
|
|
||
| singletonInstances.applyOrElse( | ||
| classType, | ||
| ct => new SimpleFileDiagnosticsSlf4jLogger(ct)) | ||
| } | ||
|
|
||
| def getLogFile(classType: Class[_]): File = { | ||
| folder.createNewFile() | ||
| val logFile = new File( | ||
| folder.getAbsolutePath() | ||
| + File.separator | ||
| + classType.getSimpleName().replace("$", "") + ".txt") | ||
|
|
||
| FileUtils.forceDeleteOnExit(logFile) | ||
| logFile | ||
| } | ||
|
|
||
| def reset(): Unit = { | ||
| try { | ||
| FileUtils.deleteDirectory(folder) | ||
| } catch { | ||
| case _: IOException => | ||
| } | ||
| } | ||
| } | ||
|
|
81 changes: 81 additions & 0 deletions
81
...ark_3-1_2-12/src/test/scala/com/azure/cosmos/spark/SimpleFileDiagnosticsSlf4jLogger.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
| package com.azure.cosmos.spark | ||
|
|
||
| import com.azure.cosmos.spark.diagnostics.SimpleDiagnosticsSlf4jLogger | ||
| import com.nimbusds.jose.util.StandardCharset | ||
|
|
||
| import java.io.{BufferedReader, File, FileOutputStream, FileReader} | ||
| import java.nio.charset.StandardCharsets | ||
| import scala.collection.mutable.ListBuffer | ||
|
|
||
| // scalastyle:off multiple.string.literals | ||
|
|
||
| private[spark] final class SimpleFileDiagnosticsSlf4jLogger(classType: Class[_]) | ||
| extends SimpleDiagnosticsSlf4jLogger(classType: Class[_]) { | ||
|
|
||
| private val thisLock = new Object() | ||
| private val file =SimpleFileDiagnosticsProvider.getLogFile(classType) | ||
|
|
||
| if (!file.exists()) { | ||
| file.createNewFile() | ||
| } | ||
|
|
||
| private val fos = new FileOutputStream(file, true) | ||
|
|
||
| override def logInfo(msg: => String): Unit = { | ||
| super.logInfo(msg) | ||
| val line = msg + "|||" | ||
| writeToFile(line) | ||
| } | ||
|
|
||
| override def logInfo(msg: => String, throwable: Throwable): Unit = { | ||
| super.logInfo(msg, throwable) | ||
| val line = if (Option.apply(throwable).isDefined) { | ||
| msg + "|||" + throwable.toString.filter(_ >= ' ') | ||
| } else { | ||
| msg + "|||" | ||
| } | ||
|
|
||
| writeToFile(line) | ||
| } | ||
|
|
||
| private def writeToFile(line: String) { | ||
| thisLock.synchronized { | ||
| val fileLock = fos.getChannel.lock() | ||
| try { | ||
| fos.write(line.getBytes(StandardCharsets.UTF_8)) | ||
| fos.write(System.getProperty("line.separator").getBytes()) | ||
| fos.flush() | ||
| fileLock.release() | ||
| } catch { | ||
| case t: Throwable => fileLock.release() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| def getMessages(): ListBuffer[(String, Option[String])] = { | ||
| val result = new ListBuffer[(String, Option[String])] | ||
| thisLock.synchronized { | ||
| val reader = new BufferedReader(new FileReader(file)) | ||
| var continue = true | ||
| while (continue) { | ||
| val line = reader.readLine() | ||
| if (line == null) { | ||
| continue = false | ||
| } else { | ||
| val columns = line.split("\\|\\|\\|") | ||
| if (columns.size == 2 && columns(1) != null && columns(1).length > 0) { | ||
| result.append((columns(0), Some(columns(1)))) | ||
| } else { | ||
| result.append((columns(0), None)) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } | ||
| result | ||
| } | ||
| } | ||
| // // scalastyle:on multiple.string.literals | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.