忽略了 JSON 阅读器中的 Spark 采样选项？

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相当于没有取样。PySparkOptionUtils 只是丢弃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

在斯卡拉：

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 中的限制会迅速升级（例如，请参阅我对Spark count vs take 和 length 的回答）并且您可以轻松地最终扫描整个输入。