You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zhengruifeng (via GitHub)" <gi...@apache.org> on 2025/11/26 12:27:29 UTC
[PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
zhengruifeng opened a new pull request, #53232:
URL: https://github.com/apache/spark/pull/53232
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
FPGrowth supports local filesystem
### Why are the changes needed?
to make FPGrowth work with local filesystem
### Does this PR introduce _any_ user-facing change?
yes, FPGrowth will work when local saving mode is one
### How was this patch tested?
updated tests
### Was this patch authored or co-authored using generative AI tooling?
no
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567012650
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
+ }
+ } else {
+ spark.read.parquet(path)
+ }
+ }
Review Comment:
`localSavingModeState` is also used **internally**, (only Spark driver code can set the flag) . Where does the doc string mentioned it ? we should remove it from doc and mark `localSavingModeState` as private field
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566936265
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
+ fileWriter.start()
+ while (batchBytesIter.hasNext) {
+ val batchBytes = batchBytesIter.next()
+ val batch = ArrowConverters.loadBatch(batchBytes, allocator)
+ loader.load(batch)
+ fileWriter.writeBatch()
+ }
+ fileWriter.close()
+ }
+}
+
+private[sql] class SparkArrowFileReader(path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val fileReader =
+ new ArrowFileReader(Files.newByteChannel(Paths.get(path)), allocator)
+
+ override def close(): Unit = {
+ allocator.close()
+ fileReader.close()
+ }
+
+ val schema: Schema = fileReader.getVectorSchemaRoot.getSchema
+
+ def read(): Iterator[Array[Byte]] = {
+ fileReader.getRecordBlocks.iterator().asScala.map { block =>
+ fileReader.loadRecordBatch(block)
+ val root = fileReader.getVectorSchemaRoot
+ val unloader = new VectorUnloader(root)
+ val batch = unloader.getRecordBatch
+ val out = new ByteArrayOutputStream()
+ val writeChannel = new WriteChannel(Channels.newChannel(out))
+ MessageSerializer.serialize(writeChannel, batch)
+ out.toByteArray
+ }
+ }
+}
+
+private[spark] object ArrowFileReadWrite {
+ def save(df: DataFrame, path: String): Unit = {
+ val maxRecordsPerBatch = df.sparkSession.sessionState.conf.arrowMaxRecordsPerBatch
+ val rdd = df.toArrowBatchRdd(maxRecordsPerBatch, "UTC", true, false)
+ val arrowSchema = ArrowUtils.toArrowSchema(df.schema, "UTC", true, false)
+ val writer = new SparkArrowFileWriter(arrowSchema, path)
+ writer.write(rdd.toLocalIterator)
Review Comment:
Instead, can we call `toLocalIterator` on original DataFrame's rdd and write rows to Arrow batches locally? Then we don't need to have the redundant Bytes?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2564895223
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
+ fileWriter.start()
+ while (batchBytesIter.hasNext) {
+ val batchBytes = batchBytesIter.next()
+ val batch = ArrowConverters.loadBatch(batchBytes, allocator)
Review Comment:
The `batch: ArrowRecordBatch` doesn't extends Serializable, so still use the `Array[Byte]` as the internal format in the PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566709758
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
Review Comment:
Below, `writer.write(rdd.toLocalIterator)` I think the code path here is to collect Arrow batches into Spark Diver, and write them in Spark Driver. So .. it should collect the Arrow batches from executors to the driver.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567228935
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
+ }
+ } else {
+ spark.read.parquet(path)
+ }
+ }
Review Comment:
Hmm, even it is just a temporary session file, is there any reason not to use Parquet but Arrow file format?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567208925
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
Review Comment:
Can we show actual session type in the error?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2570253227
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
+ fileWriter.start()
+ while (batchBytesIter.hasNext) {
+ val batchBytes = batchBytesIter.next()
+ val batch = ArrowConverters.loadBatch(batchBytes, allocator)
+ loader.load(batch)
+ fileWriter.writeBatch()
+ }
+ fileWriter.close()
+ }
+}
+
+private[sql] class SparkArrowFileReader(path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val fileReader =
+ new ArrowFileReader(Files.newByteChannel(Paths.get(path)), allocator)
+
+ override def close(): Unit = {
+ allocator.close()
+ fileReader.close()
+ }
+
+ val schema: Schema = fileReader.getVectorSchemaRoot.getSchema
+
+ def read(): Iterator[Array[Byte]] = {
+ fileReader.getRecordBlocks.iterator().asScala.map { block =>
+ fileReader.loadRecordBatch(block)
+ val root = fileReader.getVectorSchemaRoot
+ val unloader = new VectorUnloader(root)
+ val batch = unloader.getRecordBatch
+ val out = new ByteArrayOutputStream()
+ val writeChannel = new WriteChannel(Channels.newChannel(out))
+ MessageSerializer.serialize(writeChannel, batch)
+ out.toByteArray
+ }
+ }
+}
+
+private[spark] object ArrowFileReadWrite {
+ def save(df: DataFrame, path: String): Unit = {
+ val maxRecordsPerBatch = df.sparkSession.sessionState.conf.arrowMaxRecordsPerBatch
+ val rdd = df.toArrowBatchRdd(maxRecordsPerBatch, "UTC", true, false)
+ val arrowSchema = ArrowUtils.toArrowSchema(df.schema, "UTC", true, false)
+ val writer = new SparkArrowFileWriter(arrowSchema, path)
+ writer.write(rdd.toLocalIterator)
Review Comment:
we can make best of the `ArrowConverters` utils, if we use the `Bytes`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566955763
##########
mllib/src/main/scala/org/apache/spark/ml/fpm/FPGrowth.scala:
##########
@@ -343,16 +343,11 @@ object FPGrowthModel extends MLReadable[FPGrowthModel] {
class FPGrowthModelWriter(instance: FPGrowthModel) extends MLWriter {
override protected def saveImpl(path: String): Unit = {
- if (ReadWriteUtils.localSavingModeState.get()) {
- throw new UnsupportedOperationException(
- "FPGrowthModel does not support saving to local filesystem path."
- )
- }
val extraMetadata: JObject = Map("numTrainingRecords" -> instance.numTrainingRecords)
DefaultParamsWriter.saveMetadata(instance, path, sparkSession,
extraMetadata = Some(extraMetadata))
val dataPath = new Path(path, "data").toString
Review Comment:
can we pass `Path` object to `saveDataFrame` directly?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567946545
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
+ }
+ } else {
+ spark.read.parquet(path)
+ }
+ }
Review Comment:
we can read/write parquet with arrow, but it requires a new dependency
```
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-arrow</artifactId>
</dependency>
```
otherwise, I am not sure whether we have utils to read/write parquet.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566722229
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
Review Comment:
Oh I see. Okay.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2564895223
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
+ fileWriter.start()
+ while (batchBytesIter.hasNext) {
+ val batchBytes = batchBytesIter.next()
+ val batch = ArrowConverters.loadBatch(batchBytes, allocator)
Review Comment:
The `batch: ArrowRecordBatch` doesn't extends Serializable, so still use the `Array[Byte]` as the underlying data in the PR.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566992659
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
+ }
+ } else {
+ spark.read.parquet(path)
+ }
+ }
Review Comment:
hi @holdenk , as @WeichenXu123 explained https://github.com/apache/spark/pull/53150#issuecomment-3580698495, this is a runtime temporary file in spark connect server side, and will be cleaned after session close.
So I think we don't have to use a stable format here.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567017896
##########
mllib/src/main/scala/org/apache/spark/ml/fpm/FPGrowth.scala:
##########
@@ -343,16 +343,11 @@ object FPGrowthModel extends MLReadable[FPGrowthModel] {
class FPGrowthModelWriter(instance: FPGrowthModel) extends MLWriter {
override protected def saveImpl(path: String): Unit = {
- if (ReadWriteUtils.localSavingModeState.get()) {
- throw new UnsupportedOperationException(
- "FPGrowthModel does not support saving to local filesystem path."
- )
- }
val extraMetadata: JObject = Map("numTrainingRecords" -> instance.numTrainingRecords)
DefaultParamsWriter.saveMetadata(instance, path, sparkSession,
extraMetadata = Some(extraMetadata))
val dataPath = new Path(path, "data").toString
Review Comment:
sounds good!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566698276
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
Review Comment:
Dataset is already distributed on executors. Rows are written into Arrow batches in executors. If they are not to distributed again, they could be in Arrow batches, no?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566956671
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala:
##########
@@ -2340,14 +2340,13 @@ class Dataset[T] private[sql](
}
/** Convert to an RDD of serialized ArrowRecordBatches. */
- private[sql] def toArrowBatchRdd(plan: SparkPlan): RDD[Array[Byte]] = {
+ private def toArrowBatchRddImpl(
+ plan: SparkPlan,
Review Comment:
nit: 4 spaces indentation
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566710616
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
Review Comment:
I guess it's because to write down into Drivers' local file system
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "holdenk (via GitHub)" <gi...@apache.org>.
holdenk commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566922029
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
+ }
+ } else {
+ spark.read.parquet(path)
+ }
+ }
Review Comment:
So if we have `localSavingModeState` set to true this will write out an arrow file which is not stable format wise. It does look like localSavingModeState is only set to true in internal methods in Scala. Looking in the PySpark docstrings I see we tell people to use this API so I remain -0.9.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #53232:
URL: https://github.com/apache/spark/pull/53232#issuecomment-3584039414
> Due to the process of batch -> bytes -> batch -> bytes (when writing to file)
Can we have a shared util to produce RDD of arrow batches? Then we can either turn it to RDD of bytes, or write it to local files.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567663914
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
Review Comment:
make sense, will update!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][SQL][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #53232: [SPARK-54446][SQL][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format
URL: https://github.com/apache/spark/pull/53232
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566936265
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
+ fileWriter.start()
+ while (batchBytesIter.hasNext) {
+ val batchBytes = batchBytesIter.next()
+ val batch = ArrowConverters.loadBatch(batchBytes, allocator)
+ loader.load(batch)
+ fileWriter.writeBatch()
+ }
+ fileWriter.close()
+ }
+}
+
+private[sql] class SparkArrowFileReader(path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val fileReader =
+ new ArrowFileReader(Files.newByteChannel(Paths.get(path)), allocator)
+
+ override def close(): Unit = {
+ allocator.close()
+ fileReader.close()
+ }
+
+ val schema: Schema = fileReader.getVectorSchemaRoot.getSchema
+
+ def read(): Iterator[Array[Byte]] = {
+ fileReader.getRecordBlocks.iterator().asScala.map { block =>
+ fileReader.loadRecordBatch(block)
+ val root = fileReader.getVectorSchemaRoot
+ val unloader = new VectorUnloader(root)
+ val batch = unloader.getRecordBatch
+ val out = new ByteArrayOutputStream()
+ val writeChannel = new WriteChannel(Channels.newChannel(out))
+ MessageSerializer.serialize(writeChannel, batch)
+ out.toByteArray
+ }
+ }
+}
+
+private[spark] object ArrowFileReadWrite {
+ def save(df: DataFrame, path: String): Unit = {
+ val maxRecordsPerBatch = df.sparkSession.sessionState.conf.arrowMaxRecordsPerBatch
+ val rdd = df.toArrowBatchRdd(maxRecordsPerBatch, "UTC", true, false)
+ val arrowSchema = ArrowUtils.toArrowSchema(df.schema, "UTC", true, false)
+ val writer = new SparkArrowFileWriter(arrowSchema, path)
+ writer.write(rdd.toLocalIterator)
Review Comment:
Instead, can we call `toLocalIterator` on original DataFrame's rdd and write rows to Arrow batches locally?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566958078
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
Review Comment:
nit: 4 spaces indentation
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #53232:
URL: https://github.com/apache/spark/pull/53232#issuecomment-3581160516
This PR is another attempt to save ml models containing dataframes to driver's local fs.
TBH, I am not very familiar with the arrow file reader / writer
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566650682
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
Review Comment:
I think he's doing it cuz local data has to go to executors, and to do that, the arrow batches should be in ipc.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2566621669
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
Review Comment:
Where is `arrowWriter` used?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
+
+ protected val fileWriter =
+ new ArrowFileWriter(root, null, Channels.newChannel(new FileOutputStream(path)))
+
+ override def close(): Unit = {
+ root.close()
+ allocator.close()
+ fileWriter.close()
+ }
+
+ def write(batchBytesIter: Iterator[Array[Byte]]): Unit = {
Review Comment:
This looks like do such thing:
Dataset -> Arrow batches -> Bytes -> Arrow batches -> Write Arrow batches by ArrowFileWriter
Looks like the intermediate Bytes could be skipped?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567017625
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/arrow/ArrowFileReadWrite.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.arrow
+
+import java.io.{ByteArrayOutputStream, FileOutputStream}
+import java.nio.channels.Channels
+import java.nio.file.Files
+import java.nio.file.Paths
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.arrow.vector._
+import org.apache.arrow.vector.ipc.{ArrowFileReader, ArrowFileWriter, WriteChannel}
+import org.apache.arrow.vector.ipc.message.MessageSerializer
+import org.apache.arrow.vector.types.pojo.Schema
+
+import org.apache.spark.sql.classic.{DataFrame, SparkSession}
+import org.apache.spark.sql.util.ArrowUtils
+
+private[sql] class SparkArrowFileWriter(
+ arrowSchema: Schema,
+ path: String) extends AutoCloseable {
+ private val allocator =
+ ArrowUtils.rootAllocator.newChildAllocator(
+ s"to${this.getClass.getSimpleName}", 0, Long.MaxValue)
+
+ protected val root = VectorSchemaRoot.create(arrowSchema, allocator)
+ protected val loader = new VectorLoader(root)
+ protected val arrowWriter = ArrowWriter.create(root)
Review Comment:
good catch!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2567012650
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
+ }
+ } else {
+ spark.read.parquet(path)
+ }
+ }
Review Comment:
`localSavingModeState` is also used **internally**, (only Spark driver code can set the flag) . Where does the doc string mentioned it ? we should remove it from doc
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][SQL][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53232:
URL: https://github.com/apache/spark/pull/53232#discussion_r2570475992
##########
mllib/src/main/scala/org/apache/spark/ml/util/ReadWrite.scala:
##########
@@ -1142,4 +1143,31 @@ private[spark] object ReadWriteUtils {
spark.read.parquet(path).as[T].collect()
}
}
+
+ def saveDataFrame(path: String, df: DataFrame): Unit = {
+ if (localSavingModeState.get()) {
+ val filePath = Paths.get(path)
+ Files.createDirectories(filePath.getParent)
+
+ df match {
+ case d: org.apache.spark.sql.classic.DataFrame =>
+ ArrowFileReadWrite.save(d, path)
+ case _ => throw new UnsupportedOperationException("Unsupported dataframe type")
+ }
+ } else {
+ df.write.parquet(path)
+ }
+ }
+
+ def loadDataFrame(path: String, spark: SparkSession): DataFrame = {
+ if (localSavingModeState.get()) {
+ spark match {
+ case s: org.apache.spark.sql.classic.SparkSession =>
+ ArrowFileReadWrite.load(s, path)
+ case _ => throw new UnsupportedOperationException("Unsupported session type")
+ }
+ } else {
+ spark.read.parquet(path)
+ }
+ }
Review Comment:
At the end we need the in-memory data to be in arrow format, so using arrow file is more efficient.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #53232:
URL: https://github.com/apache/spark/pull/53232#issuecomment-3584125620
> Can we have a shared util to produce RDD of arrow batches? Then we can either turn it to RDD of bytes, or write it to local files.
This is actually already reusing a lot of existing utiles at `ArrowConverters.scala`. We have that same logic in Python but this `SparkArrowFileWriter` is new in JVM.
Basically `toArrowBatchRdd` is the util you meant for `batch -> bytes`.
Below code is for `bytes -> batch -> write`
```
val writer = new SparkArrowFileWriter(arrowSchema, path)
writer.write(rdd.toLocalIterator)
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54446][SQL][ML][CONNECT] FPGrowth supports local filesystem with Arrow file format [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #53232:
URL: https://github.com/apache/spark/pull/53232#issuecomment-3599782535
Merged to master.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org