1

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

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

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

我试图在加载中“不存在的文件”，我认为这是数据框定义。但我得到了以下错误。根据这本书它不应该失败吧？我肯定错过了一些东西。有人可以帮忙吗？

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 元数据？

4

2 回答 2

1

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

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

检查以下代码。

@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))
      }
于 2020-03-30T15:40:00.667 回答
1

直到这里定义了数据框并实例化了读取器对象。

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 lineage 将被创建并且在定义时不会被执行。2）当加载执行时，它将是执行时间。

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

在此处输入图像描述

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

于 2020-03-30T17:22:34.207 回答