Skip to content

[SPARK-40414][SQL][PYTHON] More generic type on PythonArrowInput and PythonArrowOutput #37864

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HeartSaVioR wants to merge 2 commits into from
Closed

[SPARK-40414][SQL][PYTHON] More generic type on PythonArrowInput and PythonArrowOutput #37864

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c800e68
[SPARK-40414][SQL][PYSPARK] More generic type on PythonArrowInput and…
HeartSaVioR Sep 13, 2022
beecc9e
reflect feedbacks
HeartSaVioR Sep 14, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
reflect feedbacks
  • Loading branch information
@HeartSaVioR
HeartSaVioR committed Sep 14, 2022
commit beecc9e66fab4ffeda2f64381297e502532bc0d8
11 changes: 4 additions & 7 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/python/PythonArrowInput.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.api.python.{BasePythonRunner, PythonRDD}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.execution.arrow.ArrowWriter
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.ArrowUtils
import org.apache.spark.util.Utils
Expand All @@ -36,8 +35,6 @@ import org.apache.spark.util.Utils
* JVM (an iterator of internal rows + additional data if required) to Python (Arrow).
*/
private[python] trait PythonArrowInput[IN] { self: BasePythonRunner[IN, _] =>
protected val sqlConf = SQLConf.get

protected val workerConf: Map[String, String]

protected val schema: StructType
Expand Down Expand Up @@ -112,10 +109,10 @@ private[python] trait BasicPythonArrowInput extends PythonArrowInput[Iterator[In
self: BasePythonRunner[Iterator[InternalRow], _] =>

protected def writeIteratorToArrowStream(
root: VectorSchemaRoot,
writer: ArrowStreamWriter,
dataOut: DataOutputStream,
inputIterator: Iterator[Iterator[InternalRow]]): Unit = {
root: VectorSchemaRoot,
writer: ArrowStreamWriter,
dataOut: DataOutputStream,
inputIterator: Iterator[Iterator[InternalRow]]): Unit = {
val arrowWriter = ArrowWriter.create(root)

while (inputIterator.hasNext) {
Expand Down
4 changes: 2 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/python/PythonArrowOutput.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,6 @@ private[python] trait BasicPythonArrowOutput extends PythonArrowOutput[ColumnarB
self: BasePythonRunner[_, ColumnarBatch] =>

protected def deserializeColumnarBatch(
batch: ColumnarBatch,
schema: StructType): ColumnarBatch = batch
batch: ColumnarBatch,
schema: StructType): ColumnarBatch = batch
}