代码之家 › 专栏 › 技术社区 › giorgionasis

intellij spark,scala读取txt文件并选择头文件并将其用作数据帧[关闭]

intellij-idea apache-spark scala

giorgionasis · 技术社区 · 6 年前

我想直接读一个文本文件到数据帧。不是文件->RDD->数据帧。有可能吗?我读了很多书,但我不能使它(读)不起作用。
阅读时,我想从中选择特定的标题。

有什么快速的解决办法吗?

我还应该进口什么?

这是我的scala文件

import org.apache.spark.{SparkConf,SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Dataset
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql._

object LoadData {


  def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Spark Job for Loading Data").setMaster("local[*]") // local[*] will access all core of your machine
val sc = new SparkContext(conf) // Create Spark Context
// Load local file data

val rdd = sc.textFile("src/main/resources/data.txt")
val df = rdd.toDF()


// Read the records

println(rdd.foreach(println))
}
}

我的SBT

 name := "HelloScala"

version := "1.0"

scalaVersion := "2.11.12"

// https://mvnrepository.com/artifact/org.apache.spark/spark-core


libraryDependencies ++= Seq(
// https://mvnrepository.com/artifact/org.apache.spark/spark-core_2.11
"org.apache.spark" %% "spark-core" % "2.3.2",
// https://mvnrepository.com/artifact/org.apache.spark/spark-sql_2.11
"org.apache.spark" %% "spark-sql" % "2.3.2"
)

我有错误错误:(16,18)值todf不是org.apache.spark.rdd.rdd[string]的成员 val df=rdd.todf()。

非常感谢你

3 回复 | 直到 6 年前

NNK 6 年前

下面是直接读取数据帧的csv文件(带标题)的示例

import org.apache.spark.sql.SparkSession

object DataFrameFromCSVFile {

  def main(args:Array[String]):Unit= {

    val spark: SparkSession = SparkSession.builder()
      .master("local[1]")
      .appName("SparkByExample")
      .getOrCreate()

    val filePath="src/main/resources/data.txt"

    //Chaining multiple options
    val df2 = spark.read.options(Map("inferSchema"->"true","sep"->",","header"->"true")).csv(filePath)
    df2.show(false)
    df2.printSchema()

  }
}

如果您有一个RDD并且想要转换为数据帧

import org.apache.spark.sql.SparkSession

object DataFrameFromRDD {

  def main(args: Array[String]): Unit = {
    val spark:SparkSession = SparkSession.builder()
      .master("local[1]")
      .appName("SparkByExample")
      .getOrCreate()

    import spark.sqlContext.implicits._
    val rdd = spark.sparkContext.parallelize(Seq(("Databricks", 20000), ("Spark", 100000), ("Hadoop", 3000)))

    val df = rdd.toDF()

    //TO-DO use df variable
  }
}

您得到的todf不是org.apache.spark.rdd.rdd[string]val df=rdd.todf()的成员,因为您没有隐式导入。

如上面的示例所示,使用import spark.sqlcontext.implicits.\并尝试rdd.todf(

快乐编码!!

谢谢

Arnon Rotem-Gal-Oz 6 年前

val df=spark.read.text("file")

你所说的特定标题是什么意思?这是CSV吗?如果是这样

val df=spark.read.format("csv").option("header", "true").load("file.csv")

然后你可以

df.select($"header1",$"header2").show() //etc.

mikeL 6 年前

您需要在设置sparkconf()之后导入spark.implicits.\uu,但应该改用sparksession。以下内容应该有效

import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder.appName(""Spark Job for Loading Data").master("local[*]").getOrCreate()
import spark.implicits._

val df=spark.read.format("csv").option("header", "true").load("data.txt")