Error while using bucket partitions

[moulimukherjee]

Seeing following error while using the bucket partition

at org.apache.iceberg.spark.source.Writer$PartitionedWriter.write(Writer.java:381)
at org.apache.iceberg.spark.source.Writer$PartitionedWriter.write(Writer.java:342)
at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$$anonfun$run$3.apply(WriteToDataSourceV2Exec.scala:118)
at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$$anonfun$run$3.apply(WriteToDataSourceV2Exec.scala:116)
at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1394)
at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.run(WriteToDataSourceV2Exec.scala:146)
at org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec$$anonfun$doExecute$2.apply(WriteToDataSourceV2Exec.scala:67)
at org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2Exec$$anonfun$doExecute$2.apply(WriteToDataSourceV2Exec.scala:66)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

The relevant code looks like

val spec = PartitionSpec.builderFor(schema).bucket("_id", 800).build()

Sorting by the column does not help as its bucketted using hash.

[rdblue]

Yeah, you need to get the bucket function to Spark and sort by that. Here's how to create a Spark UDF with the function:

import com.netflix.iceberg.transforms.Transforms
import com.netflix.iceberg.types.Types
import org.apache.spark.sql.types.IntegerType

// load the bucket transform from Iceberg to use as a UDF
val bucketTransform = Transforms.bucket[java.lang.Long](Types.LongType.get(), 16)

// needed because Scala has trouble with the Java transform type
def bucketFunc(id: Long): Int = bucketTransform.apply(id)

// create and register a UDF
val bucket16 = spark.udf.register("bucket16", bucketFunc _)

Then you can use it like this:

INSERT INTO table SELECT id, data FROM source ORDER BY bucket16(id)

[moulimukherjee]

@rdblue Thanks, let me try it out

[rdblue]

@moulimukherjee, did that solve the problem?

[moulimukherjee]

@rdblue Yes, it did. Thanks!

[github-actions]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.

[github-actions]

This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'