DataFrame Definintion 是惰性求值

Question

我是 spark 和学习它的新手。有人可以帮助解决以下问题

spark 中关于数据帧定义的引述是“一般来说，Spark 只会在作业执行时失败，而不是数据帧定义时——即使，例如，我们指向一个不存在的文件。这是由于惰性评估，"

所以我猜 spark.read.format().load() 是数据帧定义。在这个创建的数据框之上，我们应用转换和操作，如果我没记错的话，加载被读取 API 而不是转换。

我尝试 "file that does not exist" 加载，我认为这是数据帧定义。但我得到了以下错误。根据这本书应该不会失败吧？。我肯定错过了什么。有人可以帮忙吗？

df=spark.read.format('csv')
.option('header', 
'true').option('inferschema', 'true')
.load('/spark_df_data/Spark-The-Definitive-Guide/data/retail-data/by-day/2011-12-19.csv')

错误

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/hdp/current/spark2-client/python/pyspark/sql/readwriter.py", line 166, in load
    return self._df(self._jreader.load(path))
  File "/usr/hdp/current/spark2-client/python/lib/py4j-0.10.6-src.zip/py4j/java_gateway.py", line 1160, in __call__
  File "/usr/hdp/current/spark2-client/python/pyspark/sql/utils.py", line 69, in deco
    raise AnalysisException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.AnalysisException: u'Path does not exist: /spark_df_data/Spark-The-Definitive-Guide/data/retail-data/by-day/2011-12-19.csv;'

为什么数据框定义在延迟评估时引用 Hadoop 元数据？

Answer 1

Spark 是一种懒惰的进化。但是，这并不意味着它无法在加载文件时验证文件是否存在。

延迟进化发生在 DataFrame 对象上，为了创建 DataFrame 对象，他们需要首先检查文件是否存在。

勾选以下内容code。

@scala.annotation.varargs
  def load(paths: String*): DataFrame = {
    if (source.toLowerCase(Locale.ROOT) == DDLUtils.HIVE_PROVIDER) {
      throw new AnalysisException("Hive data source can only be used with tables, you can not " +
        "read files of Hive data source directly.")
    }

    DataSource.lookupDataSourceV2(source, sparkSession.sessionState.conf).map { provider =>
      val catalogManager = sparkSession.sessionState.catalogManager
      val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
        source = provider, conf = sparkSession.sessionState.conf)
      val pathsOption = if (paths.isEmpty) {
        None
      } else {
        val objectMapper = new ObjectMapper()
        Some("paths" -> objectMapper.writeValueAsString(paths.toArray))
      }

Answer 2

直到这里定义了数据框并且 reader 实例化了对象。

scala> spark.read.format("csv").option("header",true).option("inferschema",true)
res2: org.apache.spark.sql.DataFrameReader = org.apache.spark.sql.DataFrameReader@7aead157

当你真正说负载时。

res2.load('/spark_df_data/Spark-The-Definitive-Guide/data/retail-data/by-day/2011-12-19.csv') 并且文件不存在...... 是执行时间。（这意味着它必须检查数据源然后它必须从csv加载数据)

要获取数据帧，它会检查 hadoop 的元数据，因为它将检查 hdfs 此文件是否存在。

它不会那么你得到

org.apache.spark.sql.AnalysisException: Path does not exist: hdfs://203-249-241:8020/spark_df_data/Spark-The-Definitive-Guide/data/retail-data/by-day/2011-12-19.csv

一般

1) RDD/DataFrame 谱系将在定义时间创建，不会执行。 2) load执行后就是执行时间。

请参阅以下流程以更好地理解。

结论：在调用操作（执行时间以您的方式）之前，不会执行任何转换（以您的方式定义时间）

DataFrame Definintion 是惰性求值

DataFrame Definintion is lazy evaluation

dataframe

apache-spark

pyspark

rdd

lazy-evaluation