我有以下数据(只显示一个片段)
DEST_COUNTRY_NAME ORIGIN_COUNTRY_NAME count
United States Romania 15
United States Croatia 1
United States Ireland 344
Egypt United States 15
我设置为inferSchema
然后true
列describe
选择阅读。这似乎很好地工作。
scala> val data = spark.read.option("header", "true").option("inferSchema","true").csv("./data/flight-data/csv/2015-summary.csv")
scala> data.describe().show()
+-------+-----------------+-------------------+------------------+
|summary|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME| count|
+-------+-----------------+-------------------+------------------+
| count| 256| 256| 256|
| mean| null| null| 1770.765625|
| stddev| null| null|23126.516918551915|
| min| Algeria| Angola| 1|
| max| Zambia| Vietnam| 370002|
+-------+-----------------+-------------------+------------------+
如果我不指定inferSchema
,那么所有的列被视为字符串。
scala> val dataNoSchema = spark.read.option("header", "true").csv("./data/flight-data/csv/2015-summary.csv")
dataNoSchema: org.apache.spark.sql.DataFrame = [DEST_COUNTRY_NAME: string, ORIGIN_COUNTRY_NAME: string ... 1 more field]
scala> dataNoSchema.printSchema
root
|-- DEST_COUNTRY_NAME: string (nullable = true)
|-- ORIGIN_COUNTRY_NAME: string (nullable = true)
|-- count: string (nullable = true)
问题1)为什么然后Spark
给mean
和stddev
值的最后一列count
scala> dataNoSchema.describe().show();
+-------+-----------------+-------------------+------------------+
|summary|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME| count|
+-------+-----------------+-------------------+------------------+
| count| 256| 256| 256|
| mean| null| null| 1770.765625|
| stddev| null| null|23126.516918551915|
| min| Algeria| Angola| 1|
| max| Zambia| Vietnam| 986|
+-------+-----------------+-------------------+------------------+
问题2)如果现在Spark
解释count
作为numeric
柱那么为什么max
值是986,而不是37002(如在数据数据帧)
火花SQL希望成为SQL标准兼容,因此使用相同的评估规则,并且如果需要,透明胁迫类型以满足式(参见例如my answer到PySpark DataFrames - filtering using comparisons between columns of different types)。
这意味着max
和mean
/ stddev
情况是根本不相同的:
Seq.empty[String].toDF("count").agg(max("count")).explain
== Physical Plan ==
SortAggregate(key=[], functions=[max(count#69)])
+- Exchange SinglePartition
+- SortAggregate(key=[], functions=[partial_max(count#69)])
+- LocalTableScan <empty>, [count#69]
Seq.empty[String].toDF("count").agg(mean("count")).explain
== Physical Plan ==
*(2) HashAggregate(keys=[], functions=[avg(cast(count#81 as double))])
+- Exchange SinglePartition
+- *(1) HashAggregate(keys=[], functions=[partial_avg(cast(count#81 as double))])
+- LocalTableScan <empty>, [count#81].