Опции выборки Spark в JSON reader игнорируются?

Question

В следующих двух примерах количество запущенных задач и соответствующее время выполнения означают, что параметры выборки не действуют, поскольку они аналогичны заданиям, запускаемым без каких-либо параметров выборки:

val df = spark.read.options("samplingRatio",0.001).json("s3a://test/*.json.bz2")

val df = spark.read.option("sampleSize",100).json("s3a://test/*.json.bz2")

Я знаю, что явные схемы лучше всего подходят для производительности, но в удобных случаях выборка полезна.

Впервые в Spark, я неправильно использую эти опции? Попытка того же подхода в PySpark, с теми же результатами:

df = spark.read.options(samplingRatio=0.1).json("s3a://test/*.json.bz2")

df = spark.read.options(samplingRatio=None).json("s3a://test/*.json.bz2")

Answer 1

TL; DR Ни один из используемых вами вариантов не окажет существенного влияния на время выполнения:

• sampleSize не входит в число действительных JSONOptions или JSONOptionsInRead, поэтому оно будет игнорироваться.

• samplingRatio является допустимым параметром, но внутренне он использует PartitionwiseSampledRDD, поэтому процесс является линейным с точки зрения количества записей . Поэтому выборка может только снизить стоимость вывода, а не IO, что, вероятно, является здесь узким местом.

• Установка samplingRatio на None эквивалентна отсутствию выборки. PySpark OptionUtils просто отбрасывает None опции и sampleRatio по умолчанию 1.0 .

Вы можете попробовать сэмплировать данные явно. В Python

from pyspark.sql import SparkSession
from pyspark.sql.types import StructField 

def infer_json_schema(path: str, sample_size: int, **kwargs: str) -> StructType:
    spark = SparkSession.builder.getOrCreate()
    sample = spark.read.text(path).limit(sample_size).rdd.flatMap(lambda x: x)
    return spark.read.options(**kwargs).json(sample).schema

В Scala:

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StructType

def inferJsonSchema(
    path: String, sampleSize: Int, options: Map[String, String]): StructType = {
  val spark = SparkSession.builder.getOrCreate()
  val sample = spark.read.text(path).limit(sampleSize).as[String]
  spark.read.options(options).json(sample).schema
}

Помните, что для правильной работы размер выборки должен максимально соответствовать ожидаемому размеру раздела. Пределы в Spark быстро увеличиваются (см., Например, мой ответ до Счетчик искр против длины и длины ), и вы можете легко закончить сканирование всего ввода.