从 spark 数据框中访问特定行答案

【问题标题】：Access specific row from spark dataframe从 spark 数据框中访问特定行
【发布时间】：2019-10-24 21:05:05
【问题描述】：

我是 azure spark/databricks 的新手，并尝试访问特定行，例如数据框中的第 10 行。

这是我到目前为止在笔记本上所做的

1.读取表格中的 CSV 文件

spark.read
  .format("csv")
  .option("header", "true")
  .load("/mnt/training/enb/commonfiles/ramp.csv")
  .write
  .mode("overwrite")
  .saveAsTable("ramp_csv")

2。为“表”ramp_csv 创建一个 DataFrame

val rampDF = spark.read.table("ramp_csv")

3.读取特定行

我在 Scala 中使用以下逻辑

val myRow1st = rampDF.rdd.take(10).last

display(myRow1st)

它应该显示第 10 行，但我收到以下错误

command-2264596624884586:9: error: overloaded method value display with alternatives:
  [A](data: Seq[A])(implicit evidence$1: reflect.runtime.universe.TypeTag[A])Unit <and>
  (dataset: org.apache.spark.sql.Dataset[_],streamName: String,trigger: org.apache.spark.sql.streaming.Trigger,checkpointLocation: String)Unit <and>
  (model: org.apache.spark.ml.classification.DecisionTreeClassificationModel)Unit <and>
  (model: org.apache.spark.ml.regression.DecisionTreeRegressionModel)Unit <and>
  (model: org.apache.spark.ml.clustering.KMeansModel)Unit <and>
  (model: org.apache.spark.mllib.clustering.KMeansModel)Unit <and>
  (documentable: com.databricks.dbutils_v1.WithHelpMethods)Unit
 cannot be applied to (org.apache.spark.sql.Row)
display(myRow1st)
^
Command took 0.12 seconds --

您能分享一下我在这里缺少的东西吗？我尝试了一些其他的东西，但没有奏效。提前感谢您的帮助！

【问题讨论】：

标签： scala apache-spark azure-databricks

【解决方案1】：

以下是代码中发生的情况的细分：

rampDF.rdd.take(10) 返回Array[Row]

.last 返回Row

display() 接受Dataset，而您传递给它的是Row。您可以使用.show(10) 以表格形式显示前 10 行。

另一种选择是display(rampDF.limit(10))

【讨论】：

感谢您分享详细的答案。 display(rampDF.limit(10)) 工作，但它会给我前 10 行。您能否分享一下仅访问第 10 行的方法。
你这样做的方式仅适用于获取第 10 行，但这意味着将数据收集到驱动程序。如果您只想打印数据，您可以使用myRow1st.mkString(",") 获取行的字符串表示形式。这里我使用 , 作为 col 分隔符。

【解决方案2】：

我也同意 João 的回答。但是，如果您坚持将第 N 行作为 DataFrame 并避免收集到驱动程序节点（例如当 N 很大时），您可以这样做：

import org.apache.spark.sql.functions._
import spark.implicits._

val df = 1 to 100 toDF //sample data
val cols = df.columns

df
.limit(10)
.withColumn("id", monotonically_increasing_id())
.agg(max(struct(("id" +: cols).map(col(_)):_*)).alias("tenth"))
.select(cols.map(c => col("tenth."+c).alias(c)):_*)

这将返回：

+-----+
|value|
+-----+
|   10|
+-----+

【讨论】：

【解决方案3】：

我也同意 João Guitana 的回答。获得第 10 条记录的替代方法：

val df = 1 to 1000 toDF
val tenth = df.limit(10).collect.toList.last
tenth: org.apache.spark.sql.Row = [10]

这将返回 df 上的第 10 个 Row

【讨论】：