3回答

Michael_PK

2017-08-29

你仔细对比和上课的代码一致吗？肯定不是，自己仔细对比下！

0

2

Michael_PK

回复

人唯优2018

问题就出在这，你去了解下spark的运行架构，本地是不支持指定standalone模式运行的

2017-08-29

共2条回复

人唯优2018

提问者

2017-08-29

package com.immoc.spark

import org.apache.spark.rdd.RDD

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

import org.apache.spark.sql.{Row, SparkSession, types}

/**

  * DataFrame与RDD的互操作

  */

object DataFrameRDDApp {

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

    val spark = SparkSession.builder().appName("DataFrameRDDApp").master("spark://master:7077").getOrCreate()

    //inferReflection(spark)

    

    program(spark)

    spark.stop()

  }

  private def program (spark: SparkSession) = {

    val rdd = spark.sparkContext.textFile("file:///usr/local/app/spark/examples/src/main/resources/info.txt")

    val infoRDD = rdd.map(_.split(",")).map(line => Row(line(0).toInt, line(1), line(2).toInt))

    val structType = StructType(Array(StructField("id",IntegerType, true),

      StructField("name",StringType, true),

      StructField("age",IntegerType, true)

    ))

    val infoDF = spark.createDataFrame(infoRDD,structType)

    infoDF.printSchema()

    infoDF.show()

    //通过df的api进行操作

    infoDF.filter(infoDF.col("age") > 30).show

    //通过sql的方式进行操作

    infoDF.createOrReplaceTempView("infos")

    spark.sql("select * from infos where age > 30").show()

  }

  private def inferReflection(spark: SparkSession) = {

    //RDD ===> DataFrame

    val rdd = spark.sparkContext.textFile("file:///usr/local/app/spark/examples/src/main/resources/info.txt")

    //需要导入隐式转换

    import spark.implicits._

    val infoDF = rdd.map(_.split(",")).map(line => Info(line(0).toInt, line(1), line(2).toInt)).toDF()

    infoDF.show()

  }

  case class Info(id:Int, name:String, age:Int)

}

0

0

Michael_PK

2017-08-28

贴上你的代码

0

1

人唯优2018

已贴上，就是按照教程的代码

2017-08-29

共1条回复