[SPARK-39979][SQL] Add option to use large variable width vectors for arrow UDF operations by Kimahriman · Pull Request #39572 · apache/spark

Kimahriman · 2023-01-14T14:43:27Z

What changes were proposed in this pull request?

Adds a new config that uses the LargeUtf8 and LargeBinary arrow types for arrow-based UDF operations. These arrow types make arrow use LargeVarCharVector and LargeVarBinaryVector instead of the regular VarCharVector and VarBinaryVector respectively. This config is disabled by default to maintain the current behavior.

Why are the changes needed?

VarCharVector and VarBinaryVector have a size limit of 2 GiB for a single vector. This is because they use 4 byte integers to track the offsets of each value in the vector. During certain operations, it is possible to hit this limit. The most affected way that we've run into this is during a applyInPandas operation, since the entire group is sent as a single RecordBatch, and there is no way to chunk up any smaller than the entire group. However, other map and UDF operations can benefit as well without having to find the right records per batch to not hit this limit.

The large vector types use an 8 byte long to track value offsets, removing the 2 GiB total size limit.

Does this PR introduce any user-facing change?

Adds an option that can help users get around what currently results in IndexOutOfBoundsException, though this exception being raised is a bug that was fixed in Arrow and it should actually be a OversizedAllocationException in the next release which suggests using the large variable width types instead.

How was this patch tested?

A few new tests are added. I also enabled the setting by default for a full CI run and all existing tests passed. I can add more tests if needed.

Kimahriman · 2023-01-14T14:45:32Z

cc @HyukjinKwon

Wasn't sure whether to do the config based approach or just change everything to use it by default. I'm not sure if there's any other downsides other than the extra 4 bytes per value in the vector.

AmplabJenkins · 2023-01-15T00:35:35Z

Can one of the admins verify this patch?

HyukjinKwon · 2023-01-15T12:19:00Z

I like this idea. Could you have rough benchmark for this so we can assess which default value is better?

Kimahriman · 2023-01-15T14:21:05Z

Any specific way I should try to benchmark? Just using the Scala vectors directly or some mapInArrows/mapInPandas test?

Kimahriman · 2023-01-15T15:44:41Z

Threw together:

@F.udf
def func():
    return uuid.uuid4().hex

df = spark.range(67108864).select(func().alias('uuid')).repartition(1)

df.explain()

df.cache()
df.count()

def pandas_func(pdfs: Iterator[pd.DataFrame]):
    total_memory = 0
    for pdf in pdfs:
        total_memory += pdf.memory_usage(deep=True).sum()
        yield pdf
    print(f'Total memory passed to pandas: {total_memory}')

def arrow_func(batches: Iterator[pa.RecordBatch]):
    total_memory = 0
    for batch in batches:
        print(batch.schema)
        total_memory += batch.nbytes
        yield batch
    print(f'Total memory passed to arrow:  {total_memory}')

def test_pandas(df: DataFrame):
    start = datetime.now()
    df = df.mapInPandas(pandas_func, 'uuid string')
    df.count()
    print(f'Took pandas {datetime.now() - start}')

def test_arrow(df: DataFrame):
    start = datetime.now()
    df = df.mapInArrow(arrow_func, 'uuid string')
    df.count()
    print(f'Took arrow  {datetime.now() - start}')


spark.conf.set('spark.sql.execution.arrow.maxRecordsPerBatch', str(int(67108864 / 2)))
print('Regular variable width types')
test_pandas(df)
test_arrow(df)

spark.conf.set('spark.sql.execution.arrow.useLargeVarTypes', 'true')
print()
print('Large variable width types')
test_pandas(df)
test_arrow(df)

Basically creates 2 GiB of UUIDs (just wanted random strings), caches it so we don't need to worry about UUID performance, then single partition mapInPandas and mapInArrow to see speed and memory usage.

Output:

Regular variable width types
Total memory passed to pandas: 5972689152
Took pandas 0:01:04.841046
uuid: string
uuid: string
Total memory passed to arrow:  2415919104
Took arrow  0:00:18.196022

Large variable width types
Total memory passed to pandas: 5972689152
Took pandas 0:01:03.925820
uuid: large_string
uuid: large_string
Total memory passed to arrow:  2684354560
Took arrow  0:00:16.619154

Execution times seem equivalent (with large types slightly faster somehow), and memory usage is the same once it gets converted to pandas as would be expected, and is slightly higher directly in arrow also as expected. In fact that memory difference is exactly 4 extra bytes for offsets * 67108864 rows

Are you considering just using the large types everywhere by default?

Kimahriman · 2023-01-15T16:36:56Z

Also while figuring out that benchmark, I think the limit is actually 1 GiB per vector and not 2. Because Integer.MAX_VALUE = 2 GiB - 1, and the vector doubles in size whenever it needs to grow, once you go past 1 GiB it tries to double to 2 GiB which is more than 2 GiB - 1, so the doubling fails whenever it tries to grow beyond 1 GiB.

I think this is a thing introduced by apache/arrow#13815, so could potentially be considered a bug/regression for Spark with the updated Arrow version.

Edit: Confirmed. Dropped arrow version back to 7.0.0 and was able to put 67108864 - 1 hex UUIDs into a single buffer (just shy of 2 GiB) successfully, but using arrow 10.0.1 fails when trying to put 67108864 / 2 + 1 hex UUIDs into a single buffer (just over 1 GiB).

HyukjinKwon · 2023-01-16T00:55:28Z

cc @BryanCutler @viirya @ueshin FYI

Kimahriman · 2023-02-05T16:03:28Z

Created https://issues.apache.org/jira/browse/SPARK-42347 to track the arrow issue as a potential 3.4.0 blocker

q-aaronzolnailucas · 2023-05-02T16:32:39Z

Currently (before this PR) if I do a groupby -> .applyInPandas, does the 2GB limit apply to the whole group or just each column in the group?

Kimahriman · 2023-05-02T18:26:34Z

Currently (before this PR) if I do a groupby -> .applyInPandas, does the 2GB limit apply to the whole group or just each column in the group?

It's each column. And I think because of https://issues.apache.org/jira/browse/SPARK-42347 it's now limited to 1GB.

HyukjinKwon

I think this is fine. but would be great if somebody takes another look before merging it in.

HyukjinKwon

I think this is fine. but would be great if somebody takes another look before merging it in.

HyukjinKwon · 2023-05-29T00:05:01Z

Merged to master.

… function ### _Why are the changes needed?_ to adapt Spark 3.5, the new conf `spark.sql.execution.arrow.useLargeVarType` was introduced in apache/spark#39572 the signature of function `ArrowUtils#toArrowSchema` before ```scala def toArrowSchema( schema: StructType, timeZoneId: String, errorOnDuplicatedFieldNames: Boolean): Schema ``` after ```scala def toArrowSchema( schema: StructType, timeZoneId: String, errorOnDuplicatedFieldNames: Boolean, largeVarTypes: Boolean = false): Schema ``` ### _How was this patch tested?_ - [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible - [ ] Add screenshots for manual tests if appropriate - [x] [Run test](https://kyuubi.readthedocs.io/en/master/develop_tools/testing.html#running-tests) locally before make a pull request Closes #4923 from cfmcgrady/arrow-toArrowSchema. Closes #4923 3806494 [Fu Chen] Update Arguments of ArrowUtils#toArrowSchema Function Authored-by: Fu Chen <[email protected]> Signed-off-by: fwang12 <[email protected]>

… arrow UDF operations ### What changes were proposed in this pull request? Adds a new config that uses the `LargeUtf8` and `LargeBinary` arrow types for arrow-based UDF operations. These arrow types make arrow use `LargeVarCharVector` and `LargeVarBinaryVector` instead of the regular `VarCharVector` and `VarBinaryVector` respectively. This config is disabled by default to maintain the current behavior. ### Why are the changes needed? `VarCharVector` and `VarBinaryVector` have a size limit of 2 GiB for a single vector. This is because they use 4 byte integers to track the offsets of each value in the vector. During certain operations, it is possible to hit this limit. The most affected way that we've run into this is during a `applyInPandas` operation, since the entire group is sent as a single RecordBatch, and there is no way to chunk up any smaller than the entire group. However, other map and UDF operations can benefit as well without having to find the right records per batch to not hit this limit. The large vector types use an 8 byte long to track value offsets, removing the 2 GiB total size limit. ### Does this PR introduce _any_ user-facing change? Adds an option that can help users get around what currently results in `IndexOutOfBoundsException`, though this exception being raised is a bug that was fixed in Arrow and it should actually be a `OversizedAllocationException` in the next release which suggests using the large variable width types instead. ### How was this patch tested? A few new tests are added. I also enabled the setting by default for a full CI run and all existing tests passed. I can add more tests if needed. Closes apache#39572 from Kimahriman/large-binary-vector. Authored-by: Adam Binford <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]>

HyukjinKwon · 2023-06-13T07:30:38Z

+            def func(iterator):
+                for pdf in iterator:
+                    assert isinstance(pdf, pd.DataFrame)
+                    yield pdf


Hm, just noticed that we missed to fix to_arrow_type here. Meaning that JVM -> Python will use useLargeVarTypes but Python -> JVM will use use useLargeVarTypes.

For pandas related API.

For Arrow API like mapInArrow alone should be fine.

Made a followup at #41569

…eVarTypes` as an internal configuration ### What changes were proposed in this pull request? This PR is a followup of #39572 that hides the `spark.sql.execution.arrow.useLargeVarTypes` configuration as an internal configuration. ### Why are the changes needed? As described in #41569, this feature only works for `mapInArrow`, and other cases cannot be completely supported because of Arrow side limitation, see apache/arrow#35289. Therefore, this PR hides this configuration as an internal one for now. ### Does this PR introduce _any_ user-facing change? No, this configuration was not released out yet. ### How was this patch tested? Ran the Scala linter. Closes #41584 from HyukjinKwon/SPARK-39979-followup2. Authored-by: Hyukjin Kwon <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]>

…eVarTypes` as an internal configuration ### What changes were proposed in this pull request? This PR is a followup of apache#39572 that hides the `spark.sql.execution.arrow.useLargeVarTypes` configuration as an internal configuration. ### Why are the changes needed? As described in apache#41569, this feature only works for `mapInArrow`, and other cases cannot be completely supported because of Arrow side limitation, see apache/arrow#35289. Therefore, this PR hides this configuration as an internal one for now. ### Does this PR introduce _any_ user-facing change? No, this configuration was not released out yet. ### How was this patch tested? Ran the Scala linter. Closes apache#41584 from HyukjinKwon/SPARK-39979-followup2. Authored-by: Hyukjin Kwon <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]>

…ateDataFrame and toPandas with Arrow ### What changes were proposed in this pull request? This PR is a retry of #41569 that implements to use large variable types within PySpark everywhere. #39572 implemented the core logic but it only supports large variable types in the bold cases below: - `mapInArrow`: **JVM -> Python -> JVM** - Pandas UDF/Function API: **JVM -> Python** -> JVM - createDataFrame with Arrow: Python -> JVM - toPandas with Arrow: JVM -> Python This PR completes them all. ### Why are the changes needed? To consistently support the large variable types. ### Does this PR introduce _any_ user-facing change? `spark.sql.execution.arrow.useLargeVarTypes` is not released out yet so it doesn't affect any end users. ### How was this patch tested? Existing tests with `spark.sql.execution.arrow.useLargeVarTypes` enabled. Closes #49790 from HyukjinKwon/SPARK-39979-followup2. Authored-by: Hyukjin Kwon <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]>

…ateDataFrame and toPandas with Arrow ### What changes were proposed in this pull request? This PR is a retry of #41569 that implements to use large variable types within PySpark everywhere. #39572 implemented the core logic but it only supports large variable types in the bold cases below: - `mapInArrow`: **JVM -> Python -> JVM** - Pandas UDF/Function API: **JVM -> Python** -> JVM - createDataFrame with Arrow: Python -> JVM - toPandas with Arrow: JVM -> Python This PR completes them all. ### Why are the changes needed? To consistently support the large variable types. ### Does this PR introduce _any_ user-facing change? `spark.sql.execution.arrow.useLargeVarTypes` is not released out yet so it doesn't affect any end users. ### How was this patch tested? Existing tests with `spark.sql.execution.arrow.useLargeVarTypes` enabled. Closes #49790 from HyukjinKwon/SPARK-39979-followup2. Authored-by: Hyukjin Kwon <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]> (cherry picked from commit e2ef5a4) Signed-off-by: Hyukjin Kwon <[email protected]>

…ateDataFrame and toPandas with Arrow ### What changes were proposed in this pull request? This PR is a retry of apache#41569 that implements to use large variable types within PySpark everywhere. apache#39572 implemented the core logic but it only supports large variable types in the bold cases below: - `mapInArrow`: **JVM -> Python -> JVM** - Pandas UDF/Function API: **JVM -> Python** -> JVM - createDataFrame with Arrow: Python -> JVM - toPandas with Arrow: JVM -> Python This PR completes them all. ### Why are the changes needed? To consistently support the large variable types. ### Does this PR introduce _any_ user-facing change? `spark.sql.execution.arrow.useLargeVarTypes` is not released out yet so it doesn't affect any end users. ### How was this patch tested? Existing tests with `spark.sql.execution.arrow.useLargeVarTypes` enabled. Closes apache#49790 from HyukjinKwon/SPARK-39979-followup2. Authored-by: Hyukjin Kwon <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]> (cherry picked from commit 454111a) Signed-off-by: Hyukjin Kwon <[email protected]>

github-actions Bot added CORE PYTHON SQL labels Jan 14, 2023

Kimahriman mentioned this pull request Jan 15, 2023

ARROW-17338: [Java] The maximum request memory of BaseVariableWidthVector should limit to Integer.MAX_VALUE apache/arrow#13815

Merged

Kimahriman force-pushed the large-binary-vector branch from bf140c0 to ebb7db5 Compare February 4, 2023 13:39

Kimahriman mentioned this pull request Feb 4, 2023

[Java] BaseVariableWidthVector only supports 1 GiB through safe interfaces apache/arrow-java#220

Open

Kimahriman force-pushed the large-binary-vector branch from ebb7db5 to 215bc7f Compare March 30, 2023 11:48

Kimahriman force-pushed the large-binary-vector branch 3 times, most recently from 215bc7f to 96b8892 Compare April 29, 2023 13:30

Kimahriman added 4 commits May 6, 2023 10:35

Implement for all udf/map functions

384eebc

Fix formatting and add arrow map test

aec1981

Disable by default and remove commented out line in test

8e32c0c

Fix arrow test and reformat some scala

48a0af5

Kimahriman force-pushed the large-binary-vector branch from 96b8892 to 48a0af5 Compare May 6, 2023 14:36

HyukjinKwon reviewed May 24, 2023

View reviewed changes

Comment thread sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala Resolved

HyukjinKwon approved these changes May 24, 2023

View reviewed changes

Update version for conf

229c6e2

HyukjinKwon closed this in 9cea0fb May 29, 2023

cfmcgrady mentioned this pull request Jun 5, 2023

[ARROW] Update arguments of ArrowUtils#toArrowSchema function apache/kyuubi#4923

Closed

1 of 3 tasks

HyukjinKwon reviewed Jun 13, 2023

View reviewed changes

This was referenced Jun 13, 2023

[SPARK-39979][SQL][FOLLOW-UP] Support large variable types in pandas UDF, createDataFrame and toPandas with Arrow #41569

Closed

[SPARK-39979][SQL][FOLLOW-UP] Make spark.sql.execution.arrow.useLargeVarTypes as an internal configuration #41584

Closed

HyukjinKwon mentioned this pull request Jan 8, 2025

[SPARK-50765][PYTHON] Remove invalid test for spark.sql.execution.arrow.useLargeVarTypes #49408

Closed

HyukjinKwon mentioned this pull request Feb 4, 2025

[SPARK-51079][PYTHON] Support large variable types in pandas UDF, createDataFrame and toPandas with Arrow #49790

Closed

Conversation

Kimahriman commented Jan 14, 2023

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Kimahriman commented Jan 14, 2023

Uh oh!

AmplabJenkins commented Jan 15, 2023

Uh oh!

HyukjinKwon commented Jan 15, 2023

Uh oh!

Kimahriman commented Jan 15, 2023

Uh oh!

Kimahriman commented Jan 15, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Kimahriman commented Jan 15, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HyukjinKwon commented Jan 16, 2023

Uh oh!

Kimahriman commented Feb 5, 2023

Uh oh!

q-aaronzolnailucas commented May 2, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Kimahriman commented May 2, 2023

Uh oh!

Uh oh!

HyukjinKwon left a comment

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon left a comment

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon commented May 29, 2023

Uh oh!

HyukjinKwon Jun 13, 2023

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon Jun 13, 2023

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon Jun 13, 2023

Choose a reason for hiding this comment

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Kimahriman commented Jan 15, 2023 •

edited

Loading

Kimahriman commented Jan 15, 2023 •

edited

Loading

q-aaronzolnailucas commented May 2, 2023 •

edited

Loading