This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 6051232a feat: Add specific configs for converting Spark Parquet and
JSON data to Arrow (#832)
6051232a is described below
commit 6051232ae76f7dd0b7797f8bb23a4133332768e4
Author: Andy Grove <[email protected]>
AuthorDate: Fri Aug 16 06:46:17 2024 -0600
feat: Add specific configs for converting Spark Parquet and JSON data to
Arrow (#832)
* Add option to enable JSON scan
* support v1 and v2 data sources
* formatting
* format
* Add schema check
* fix regression
* improve configs and docs
* fix
* improve test and fix typo in doc
* renaming some variables but no change to public config names
* improve test
* improve test
* Update
spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
Co-authored-by: Liang-Chi Hsieh <[email protected]>
* Update
spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
Co-authored-by: Liang-Chi Hsieh <[email protected]>
* improve config description
* fix path
---------
Co-authored-by: Liang-Chi Hsieh <[email protected]>
---
.../main/scala/org/apache/comet/CometConf.scala | 38 +++++++++++++-----
docs/source/index.rst | 5 ++-
docs/source/user-guide/configs.md | 6 ++-
docs/source/user-guide/datasources.md | 32 +++++++++++++++
docs/source/user-guide/operators.md | 27 +++++++------
.../apache/comet/CometSparkSessionExtensions.scala | 39 ++++++++++++++++---
.../test/resources/test-data/json-test-1.ndjson | 4 ++
.../org/apache/comet/CometExpressionSuite.scala | 45 ++++++++++++----------
.../comet/exec/CometColumnarShuffleSuite.scala | 2 +-
.../org/apache/comet/exec/CometExecSuite.scala | 14 +++++++
.../scala/org/apache/spark/sql/CometTestBase.scala | 2 +-
11 files changed, 158 insertions(+), 56 deletions(-)
diff --git a/common/src/main/scala/org/apache/comet/CometConf.scala
b/common/src/main/scala/org/apache/comet/CometConf.scala
index 1db80298..324c1d02 100644
--- a/common/src/main/scala/org/apache/comet/CometConf.scala
+++ b/common/src/main/scala/org/apache/comet/CometConf.scala
@@ -84,15 +84,33 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(sys.env.getOrElse("ENABLE_COMET", "true").toBoolean)
- val COMET_SCAN_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.scan.enabled")
+ val COMET_NATIVE_SCAN_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.scan.enabled")
.doc(
- "Whether to enable Comet scan. When this is turned on, Spark will use
Comet to read " +
- "Parquet data source. Note that to enable native vectorized execution,
both this " +
- "config and 'spark.comet.exec.enabled' need to be enabled. By default,
this config " +
- "is true.")
+ "Whether to enable native scans. When this is turned on, Spark will use
Comet to " +
+ "read supported data sources (currently only Parquet is supported
natively). Note " +
+ "that to enable native vectorized execution, both this config and " +
+ "'spark.comet.exec.enabled' need to be enabled. By default, this
config is true.")
.booleanConf
.createWithDefault(true)
+ val COMET_CONVERT_FROM_PARQUET_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.parquet.enabled")
+ .doc(
+ "When enabled, data from Spark (non-native) Parquet v1 and v2 scans
will be converted to " +
+ "Arrow format. Note that to enable native vectorized execution, both
this config and " +
+ "'spark.comet.exec.enabled' need to be enabled.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val COMET_CONVERT_FROM_JSON_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.convert.json.enabled")
+ .doc(
+ "When enabled, data from Spark (non-native) JSON v1 and v2 scans will
be converted to " +
+ "Arrow format. Note that to enable native vectorized execution, both
this config and " +
+ "'spark.comet.exec.enabled' need to be enabled.")
+ .booleanConf
+ .createWithDefault(false)
+
val COMET_EXEC_ENABLED: ConfigEntry[Boolean] =
conf(s"$COMET_EXEC_CONFIG_PREFIX.enabled")
.doc(
"Whether to enable Comet native vectorized execution for Spark. This
controls whether " +
@@ -457,20 +475,20 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(COMET_SCHEMA_EVOLUTION_ENABLED_DEFAULT)
- val COMET_SPARK_TO_COLUMNAR_ENABLED: ConfigEntry[Boolean] =
+ val COMET_SPARK_TO_ARROW_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.sparkToColumnar.enabled")
.internal()
- .doc("Whether to enable Spark to Comet columnar conversion. When this is
turned on, " +
+ .doc("Whether to enable Spark to Arrow columnar conversion. When this is
turned on, " +
"Comet will convert operators in " +
- "`spark.comet.sparkToColumnar.supportedOperatorList` into Comet
columnar based before " +
+ "`spark.comet.sparkToColumnar.supportedOperatorList` into Arrow
columnar format before " +
"processing.")
.booleanConf
.createWithDefault(false)
- val COMET_SPARK_TO_COLUMNAR_SUPPORTED_OPERATOR_LIST:
ConfigEntry[Seq[String]] =
+ val COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST: ConfigEntry[Seq[String]] =
conf("spark.comet.sparkToColumnar.supportedOperatorList")
.doc(
- "A comma-separated list of operators that will be converted to Comet
columnar " +
+ "A comma-separated list of operators that will be converted to Arrow
columnar " +
"format when 'spark.comet.sparkToColumnar.enabled' is true")
.stringConf
.toSequence
diff --git a/docs/source/index.rst b/docs/source/index.rst
index 101c5785..4bf5d9fd 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -42,9 +42,10 @@ as a native runtime to achieve improvement in terms of query
efficiency and quer
Comet Overview <user-guide/overview>
Installing Comet <user-guide/installation>
- Supported Expressions <user-guide/expressions>
- Supported Operators <user-guide/operators>
+ Supported Data Sources <user-guide/datasources>
Supported Data Types <user-guide/datatypes>
+ Supported Operators <user-guide/operators>
+ Supported Expressions <user-guide/expressions>
Configuration Settings <user-guide/configs>
Compatibility Guide <user-guide/compatibility>
Tuning Guide <user-guide/tuning>
diff --git a/docs/source/user-guide/configs.md
b/docs/source/user-guide/configs.md
index 836f6444..8465baa7 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -30,6 +30,8 @@ Comet provides the following configuration settings.
| spark.comet.columnar.shuffle.async.max.thread.num | Maximum number of
threads on an executor used for Comet async columnar shuffle. By default, this
config is 100. This is the upper bound of total number of shuffle threads per
executor. In other words, if the number of cores * the number of shuffle
threads per task `spark.comet.columnar.shuffle.async.thread.num` is larger than
this config. Comet will use this config as the number of shuffle threads per
executor instead. | 100 |
| spark.comet.columnar.shuffle.async.thread.num | Number of threads used for
Comet async columnar shuffle per shuffle task. By default, this config is 3.
Note that more threads means more memory requirement to buffer shuffle data
before flushing to disk. Also, more threads may not always improve performance,
and should be set based on the number of cores available. | 3 |
| spark.comet.columnar.shuffle.memory.factor | Fraction of Comet memory to be
allocated per executor process for Comet shuffle. Comet memory size is
specified by `spark.comet.memoryOverhead` or calculated by
`spark.comet.memory.overhead.factor` * `spark.executor.memory`. By default,
this config is 1.0. | 1.0 |
+| spark.comet.convert.json.enabled | When enabled, data from Spark
(non-native) JSON v1 and v2 scans will be converted to Arrow format. Note that
to enable native vectorized execution, both this config and
'spark.comet.exec.enabled' need to be enabled. | false |
+| spark.comet.convert.parquet.enabled | When enabled, data from Spark
(non-native) Parquet v1 and v2 scans will be converted to Arrow format. Note
that to enable native vectorized execution, both this config and
'spark.comet.exec.enabled' need to be enabled. | false |
| spark.comet.debug.enabled | Whether to enable debug mode for Comet. By
default, this config is false. When enabled, Comet will do additional checks
for debugging purpose. For example, validating array when importing arrays from
JVM at native side. Note that these checks may be expensive in performance and
should only be enabled for debugging purpose. | false |
| spark.comet.enabled | Whether to enable Comet extension for Spark. When this
is turned on, Spark will use Comet to read Parquet data source. Note that to
enable native vectorized execution, both this config and
'spark.comet.exec.enabled' need to be enabled. By default, this config is the
value of the env var `ENABLE_COMET` if set, or true otherwise. | true |
| spark.comet.exceptionOnDatetimeRebase | Whether to throw exception when
seeing dates/timestamps from the legacy hybrid (Julian + Gregorian) calendar.
Since Spark 3, dates/timestamps were written according to the Proleptic
Gregorian calendar. When this is true, Comet will throw exceptions when seeing
these dates/timestamps that were written by Spark version before 3.0. If this
is false, these dates/timestamps will be read as if they were written to the
Proleptic Gregorian calendar and w [...]
@@ -64,8 +66,8 @@ Comet provides the following configuration settings.
| spark.comet.nativeLoadRequired | Whether to require Comet native library to
load successfully when Comet is enabled. If not, Comet will silently fallback
to Spark when it fails to load the native lib. Otherwise, an error will be
thrown and the Spark job will be aborted. | false |
| spark.comet.parquet.enable.directBuffer | Whether to use Java direct byte
buffer when reading Parquet. By default, this is false | false |
| spark.comet.regexp.allowIncompatible | Comet is not currently fully
compatible with Spark for all regular expressions. Set this config to true to
allow them anyway using Rust's regular expression engine. See compatibility
guide for more information. | false |
-| spark.comet.scan.enabled | Whether to enable Comet scan. When this is turned
on, Spark will use Comet to read Parquet data source. Note that to enable
native vectorized execution, both this config and 'spark.comet.exec.enabled'
need to be enabled. By default, this config is true. | true |
+| spark.comet.scan.enabled | Whether to enable native scans. When this is
turned on, Spark will use Comet to read supported data sources (currently only
Parquet is supported natively). Note that to enable native vectorized
execution, both this config and 'spark.comet.exec.enabled' need to be enabled.
By default, this config is true. | true |
| spark.comet.scan.preFetch.enabled | Whether to enable pre-fetching feature
of CometScan. By default is disabled. | false |
| spark.comet.scan.preFetch.threadNum | The number of threads running
pre-fetching for CometScan. Effective if spark.comet.scan.preFetch.enabled is
enabled. By default it is 2. Note that more pre-fetching threads means more
memory requirement to store pre-fetched row groups. | 2 |
| spark.comet.shuffle.preferDictionary.ratio | The ratio of total values to
distinct values in a string column to decide whether to prefer dictionary
encoding when shuffling the column. If the ratio is higher than this config,
dictionary encoding will be used on shuffling string column. This config is
effective if it is higher than 1.0. By default, this config is 10.0. Note that
this config is only used when `spark.comet.exec.shuffle.mode` is `jvm`. | 10.0 |
-| spark.comet.sparkToColumnar.supportedOperatorList | A comma-separated list
of operators that will be converted to Comet columnar format when
'spark.comet.sparkToColumnar.enabled' is true | Range,InMemoryTableScan |
+| spark.comet.sparkToColumnar.supportedOperatorList | A comma-separated list
of operators that will be converted to Arrow columnar format when
'spark.comet.sparkToColumnar.enabled' is true | Range,InMemoryTableScan |
diff --git a/docs/source/user-guide/datasources.md
b/docs/source/user-guide/datasources.md
new file mode 100644
index 00000000..92c13f67
--- /dev/null
+++ b/docs/source/user-guide/datasources.md
@@ -0,0 +1,32 @@
+<!---
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# Supported Spark Data Sources
+
+## Parquet
+
+When `spark.comet.scan.enabled` is enabled, Parquet scans will be performed
natively by Comet if all data types
+in the schema are supported. When this option is not enabled, the scan will
fall back to Spark. In this case,
+enabling `spark.comet.convert.parquet.enabled` will immediately convert the
data into Arrow format, allowing native
+execution to happen after that, but the process may not be efficient.
+
+## JSON
+
+Comet does not provide native JSON scan, but when
`spark.comet.convert.json.enabled` is enabled, data is immediately
+converted into Arrow format, allowing native execution to happen after that.
diff --git a/docs/source/user-guide/operators.md
b/docs/source/user-guide/operators.md
index e3a3ac52..7edbb1e7 100644
--- a/docs/source/user-guide/operators.md
+++ b/docs/source/user-guide/operators.md
@@ -22,17 +22,16 @@
The following Spark operators are currently replaced with native versions.
Query stages that contain any operators
not supported by Comet will fall back to regular Spark execution.
-| Operator | Notes |
-| -------------------------------------------- | ----- |
-| FileSourceScanExec/BatchScanExec for Parquet | |
-| Projection | |
-| Filter | |
-| Sort | |
-| Hash Aggregate | |
-| Limit | |
-| Sort-merge Join | |
-| Hash Join | |
-| BroadcastHashJoinExec | |
-| Shuffle | |
-| Expand | |
-| Union | |
+| Operator | Notes |
+| --------------------- | ----- |
+| Projection | |
+| Filter | |
+| Sort | |
+| Hash Aggregate | |
+| Limit | |
+| Sort-merge Join | |
+| Hash Join | |
+| BroadcastHashJoinExec | |
+| Shuffle | |
+| Expand | |
+| Union | |
diff --git
a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
index f24624dc..e978e7d3 100644
--- a/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
+++ b/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
@@ -35,8 +35,10 @@ import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.adaptive.{AQEShuffleReadExec,
BroadcastQueryStageExec, ShuffleQueryStageExec}
import org.apache.spark.sql.execution.aggregate.HashAggregateExec
import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.execution.datasources.json.JsonFileFormat
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.v2.json.JsonScan
import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
ReusedExchangeExec, ShuffleExchangeExec}
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec,
ShuffledHashJoinExec, SortMergeJoinExec}
@@ -1095,7 +1097,7 @@ object CometSparkSessionExtensions extends Logging {
}
private[comet] def isCometScanEnabled(conf: SQLConf): Boolean = {
- COMET_SCAN_ENABLED.get(conf)
+ COMET_NATIVE_SCAN_ENABLED.get(conf)
}
private[comet] def isCometExecEnabled(conf: SQLConf): Boolean = {
@@ -1131,12 +1133,39 @@ object CometSparkSessionExtensions extends Logging {
// operators can have a chance to be converted to columnar. Leaf operators
that output
// columnar batches, such as Spark's vectorized readers, will also be
converted to native
// comet batches.
- // TODO: consider converting other intermediate operators to columnar.
- op.isInstanceOf[LeafExecNode] &&
CometSparkToColumnarExec.isSchemaSupported(op.schema) &&
- COMET_SPARK_TO_COLUMNAR_ENABLED.get(conf) && {
+ if (CometSparkToColumnarExec.isSchemaSupported(op.schema)) {
+ op match {
+ // Convert Spark DS v1 scan to Arrow format
+ case scan: FileSourceScanExec =>
+ scan.relation.fileFormat match {
+ case _: JsonFileFormat =>
CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf)
+ case _: ParquetFileFormat =>
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf)
+ case _ => isSparkToArrowEnabled(conf, op)
+ }
+ // Convert Spark DS v2 scan to Arrow format
+ case scan: BatchScanExec =>
+ scan.scan match {
+ case _: JsonScan =>
CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf)
+ case _: ParquetScan =>
CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf)
+ case _ => isSparkToArrowEnabled(conf, op)
+ }
+ // other leaf nodes
+ case _: LeafExecNode =>
+ isSparkToArrowEnabled(conf, op)
+ case _ =>
+ // TODO: consider converting other intermediate operators to
columnar.
+ false
+ }
+ } else {
+ false
+ }
+ }
+
+ private def isSparkToArrowEnabled(conf: SQLConf, op: SparkPlan) = {
+ COMET_SPARK_TO_ARROW_ENABLED.get(conf) && {
val simpleClassName = Utils.getSimpleName(op.getClass)
val nodeName = simpleClassName.replaceAll("Exec$", "")
-
COMET_SPARK_TO_COLUMNAR_SUPPORTED_OPERATOR_LIST.get(conf).contains(nodeName)
+ COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.get(conf).contains(nodeName)
}
}
diff --git a/spark/src/test/resources/test-data/json-test-1.ndjson
b/spark/src/test/resources/test-data/json-test-1.ndjson
new file mode 100644
index 00000000..324b3060
--- /dev/null
+++ b/spark/src/test/resources/test-data/json-test-1.ndjson
@@ -0,0 +1,4 @@
+{ "a": [1], "b": { "c": "foo" }}
+{ "a": 2, "b": { "d": "bar" }}
+{ "b": { "d": "bar" }}
+{ "a": 3 }
diff --git a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
index b9c5c3ca..d27c1791 100644
--- a/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala
@@ -1974,27 +1974,30 @@ class CometExpressionSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
test("get_struct_field") {
- withSQLConf(
- CometConf.COMET_SPARK_TO_COLUMNAR_ENABLED.key -> "true",
- CometConf.COMET_SPARK_TO_COLUMNAR_SUPPORTED_OPERATOR_LIST.key ->
"FileSourceScan") {
- withTempPath { dir =>
- var df = spark
- .range(5)
- // Add both a null struct and null inner value
- .select(
- when(
- col("id") > 1,
- struct(
- when(col("id") > 2, col("id")).alias("id"),
- when(col("id") > 2, struct(when(col("id") > 3,
col("id")).alias("id")))
- .as("nested2")))
- .alias("nested1"))
-
- df.write.parquet(dir.toString())
-
- df = spark.read.parquet(dir.toString())
- checkSparkAnswerAndOperator(df.select("nested1.id"))
- checkSparkAnswerAndOperator(df.select("nested1.nested2.id"))
+ Seq("", "parquet").foreach { v1List =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
+ CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false",
+ CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.key -> "true") {
+ withTempPath { dir =>
+ var df = spark
+ .range(5)
+ // Add both a null struct and null inner value
+ .select(
+ when(
+ col("id") > 1,
+ struct(
+ when(col("id") > 2, col("id")).alias("id"),
+ when(col("id") > 2, struct(when(col("id") > 3,
col("id")).alias("id")))
+ .as("nested2")))
+ .alias("nested1"))
+
+ df.write.parquet(dir.toString())
+
+ df = spark.read.parquet(dir.toString())
+ checkSparkAnswerAndOperator(df.select("nested1.id"))
+ checkSparkAnswerAndOperator(df.select("nested1.nested2.id"))
+ }
}
}
}
diff --git
a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
index 78b4bbb9..a2f07155 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala
@@ -831,7 +831,7 @@ abstract class CometColumnarShuffleSuite extends
CometTestBase with AdaptiveSpar
withSQLConf(
SQLConf.USE_V1_SOURCE_LIST.key -> "", // Use DataSourceV2
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", // Disable AQE
- CometConf.COMET_SCAN_ENABLED.key -> "false") { // Disable CometScan to
use Spark BatchScan
+ CometConf.COMET_NATIVE_SCAN_ENABLED.key -> "false") { // Disable
CometScan to use Spark BatchScan
withParquetTable((0 until 5).map(i => (i, (i + 1).toLong)), "tbl") {
val df = sql("SELECT * FROM tbl")
val shuffled = df
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
index 5ca9401c..15864a33 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
@@ -1606,6 +1606,20 @@ class CometExecSuite extends CometTestBase {
}
})
}
+
+ test("read JSON file") {
+ Seq("", "json").foreach { v1List =>
+ withSQLConf(
+ SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
+ CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "true",
+ CometConf.COMET_CONVERT_FROM_JSON_ENABLED.key -> "true") {
+ spark.read
+ .json("src/test/resources/test-data/json-test-1.ndjson")
+ .createOrReplaceTempView("tbl")
+ checkSparkAnswerAndOperator("SELECT a, b.c, b.d FROM tbl")
+ }
+ }
+ }
}
case class BucketedTableTestSpec(
diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
index fd0328c8..a88f9f45 100644
--- a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
+++ b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala
@@ -79,7 +79,7 @@ abstract class CometTestBase
conf.set(CometConf.COMET_ENABLED.key, "true")
conf.set(CometConf.COMET_EXEC_ENABLED.key, "true")
conf.set(CometConf.COMET_EXEC_ALL_OPERATOR_ENABLED.key, "true")
- conf.set(CometConf.COMET_SPARK_TO_COLUMNAR_ENABLED.key, "true")
+ conf.set(CometConf.COMET_SPARK_TO_ARROW_ENABLED.key, "true")
conf.set(CometConf.COMET_MEMORY_OVERHEAD.key, "2g")
conf
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
