在spark中创建行组大小小于100的拼花地板文件

我有一个有少量字段的spark数据框。一些字段是巨大的二进制Blob。整行的大小约为50 MB。

我正在将数据帧保存为拼花格式。我正在使用控制行组的大小 parquet.block.size 参数

Spark将生成一个拼花地板文件,但我总是在一个行组中获得至少100行。这对我来说是一个问题,因为块大小可能会变成千兆字节,这在我的应用程序中无法正常工作。

拼花地板块大小 只要大小足够容纳100多行,就可以正常工作。

我修改了 InternalParquetRecordWriter.java 成为 MINIMUM_RECORD_COUNT_FOR_CHECK = 2 ,修复了该问题,但是,我找不到支持调整此硬编码常量的配置值。

是否有其他/更好的方法可以获得小于100的行组大小?

这是我的代码片段:

from pyspark import Row
from pyspark.sql import SparkSession
import numpy as np

from pyspark.sql.types import StructType, StructField, BinaryType


def fake_row(x):
    result = bytearray(np.random.randint(0, 127, (3 * 1024 * 1024 / 2), dtype=np.uint8).tobytes())
    return Row(result, result)

spark_session = SparkSession \
    .builder \
    .appName("bbox2d_dataset_extraction") \
    .config("spark.driver.memory", "12g") \
    .config("spark.executor.memory", "4g")

spark_session.master('local[5]')

spark = spark_session.getOrCreate()
sc = spark.sparkContext
sc._jsc.hadoopConfiguration().setInt("parquet.block.size", 8 * 1024 * 1024)

index = sc.parallelize(range(50), 5)
huge_rows = index.map(fake_row)
schema = StructType([StructField('f1', BinaryType(), False), StructField('f2', BinaryType(), False)])

bbox2d_dataframe = spark.createDataFrame(huge_rows, schema).coalesce(1)
bbox2d_dataframe. \
    write.option("compression", "none"). \
    mode('overwrite'). \
    parquet('/tmp/huge/')