zhztheplayer commented on code in PR #6855:
URL: https://github.com/apache/incubator-gluten/pull/6855#discussion_r1719173051
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala:
##########
@@ -27,138 +26,67 @@ import org.apache.gluten.vectorized.{JniLibLoader,
JniWorkspace}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.api.plugin.PluginContext
+import org.apache.spark.internal.Logging
import
org.apache.spark.sql.execution.datasources.velox.{VeloxOrcWriterInjects,
VeloxParquetWriterInjects, VeloxRowSplitter}
import org.apache.spark.sql.expression.UDFResolver
-import org.apache.spark.sql.internal.{GlutenConfigUtil, StaticSQLConf}
-import org.apache.spark.util.SparkDirectoryUtil
+import org.apache.spark.sql.internal.GlutenConfigUtil
+import org.apache.spark.util.{SparkDirectoryUtil, SparkResourceUtil}
import org.apache.commons.lang3.StringUtils
-import scala.sys.process._
+import java.util.concurrent.atomic.AtomicBoolean
-class VeloxListenerApi extends ListenerApi {
- private val ARROW_VERSION = "1500"
+class VeloxListenerApi extends ListenerApi with Logging {
+ import VeloxListenerApi._
override def onDriverStart(sc: SparkContext, pc: PluginContext): Unit = {
- val conf = pc.conf()
- // sql table cache serializer
- if (conf.getBoolean(GlutenConfig.COLUMNAR_TABLE_CACHE_ENABLED.key,
defaultValue = false)) {
- conf.set(
- StaticSQLConf.SPARK_CACHE_SERIALIZER.key,
- "org.apache.spark.sql.execution.ColumnarCachedBatchSerializer")
+ if (!driverInitialized.compareAndSet(false, true)) {
+ // Make sure we call the static initializers only once.
+ logInfo(
+ "Skip rerunning static initializers since they are only supposed to
run once." +
+ " You see this message probably because you are creating a new
SparkSession.")
+ return
}
- initialize(conf, isDriver = true)
+
+ // Static initializers for driver.
+ val conf = pc.conf()
+ SparkDirectoryUtil.init(conf)
+ UDFResolver.resolveUdfConf(conf, isDriver = true)
+ initialize(conf)
}
override def onDriverShutdown(): Unit = shutdown()
override def onExecutorStart(pc: PluginContext): Unit = {
- initialize(pc.conf(), isDriver = false)
- }
-
- override def onExecutorShutdown(): Unit = shutdown()
-
- private def getLibraryLoaderForOS(
- systemName: String,
- systemVersion: String,
- system: String): SharedLibraryLoader = {
- if (systemName.contains("Ubuntu") && systemVersion.startsWith("20.04")) {
- new SharedLibraryLoaderUbuntu2004
- } else if (systemName.contains("Ubuntu") &&
systemVersion.startsWith("22.04")) {
- new SharedLibraryLoaderUbuntu2204
- } else if (systemName.contains("CentOS") && systemVersion.startsWith("9"))
{
- new SharedLibraryLoaderCentos9
- } else if (systemName.contains("CentOS") && systemVersion.startsWith("8"))
{
- new SharedLibraryLoaderCentos8
- } else if (systemName.contains("CentOS") && systemVersion.startsWith("7"))
{
- new SharedLibraryLoaderCentos7
- } else if (systemName.contains("Alibaba Cloud Linux") &&
systemVersion.startsWith("3")) {
- new SharedLibraryLoaderCentos8
- } else if (systemName.contains("Alibaba Cloud Linux") &&
systemVersion.startsWith("2")) {
- new SharedLibraryLoaderCentos7
- } else if (systemName.contains("Anolis") && systemVersion.startsWith("8"))
{
- new SharedLibraryLoaderCentos8
- } else if (systemName.contains("Anolis") && systemVersion.startsWith("7"))
{
- new SharedLibraryLoaderCentos7
- } else if (system.contains("tencentos") && system.contains("2.4")) {
- new SharedLibraryLoaderCentos7
- } else if (system.contains("tencentos") && system.contains("3.2")) {
- new SharedLibraryLoaderCentos8
- } else if (systemName.contains("Red Hat") &&
systemVersion.startsWith("9")) {
- new SharedLibraryLoaderCentos9
- } else if (systemName.contains("Red Hat") &&
systemVersion.startsWith("8")) {
- new SharedLibraryLoaderCentos8
- } else if (systemName.contains("Red Hat") &&
systemVersion.startsWith("7")) {
- new SharedLibraryLoaderCentos7
- } else if (systemName.contains("Debian") &&
systemVersion.startsWith("11")) {
- new SharedLibraryLoaderDebian11
- } else if (systemName.contains("Debian") &&
systemVersion.startsWith("12")) {
- new SharedLibraryLoaderDebian12
- } else {
- throw new GlutenException(
- s"Found unsupported OS($systemName, $systemVersion)! Currently,
Gluten's Velox backend" +
- " only supports Ubuntu 20.04/22.04, CentOS 7/8, " +
- "Alibaba Cloud Linux 2/3 & Anolis 7/8, tencentos 2.4/3.2, RedHat
7/8, " +
- "Debian 11/12.")
+ if (!executorInitialized.compareAndSet(false, true)) {
+ // Make sure we call the static initializers only once.
+ logInfo(
+ "Skip rerunning static initializers since they are only supposed to
run once." +
+ " You see this message probably because you are creating a new
SparkSession.")
+ return
}
- }
-
- private def loadLibFromJar(load: JniLibLoader, conf: SparkConf): Unit = {
- val systemName = conf.getOption(GlutenConfig.GLUTEN_LOAD_LIB_OS)
- val loader = if (systemName.isDefined) {
- val systemVersion =
conf.getOption(GlutenConfig.GLUTEN_LOAD_LIB_OS_VERSION)
- if (systemVersion.isEmpty) {
- throw new GlutenException(
- s"${GlutenConfig.GLUTEN_LOAD_LIB_OS_VERSION} must be specified when
specifies the " +
- s"${GlutenConfig.GLUTEN_LOAD_LIB_OS}")
- }
- getLibraryLoaderForOS(systemName.get, systemVersion.get, "")
- } else {
- val system = "cat /etc/os-release".!!
- val systemNamePattern = "^NAME=\"?(.*)\"?".r
- val systemVersionPattern = "^VERSION=\"?(.*)\"?".r
- val systemInfoLines = system.stripMargin.split("\n")
- val systemNamePattern(systemName) =
- systemInfoLines.find(_.startsWith("NAME=")).getOrElse("")
- val systemVersionPattern(systemVersion) =
- systemInfoLines.find(_.startsWith("VERSION=")).getOrElse("")
- if (systemName.isEmpty || systemVersion.isEmpty) {
- throw new GlutenException("Failed to get OS name and version info.")
- }
- getLibraryLoaderForOS(systemName, systemVersion, system)
+ val conf = pc.conf
+ if (inLocalMode(conf)) {
Review Comment:
I am OK to both... Keeping it would shorten the calling code a little bit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
