caroljmcdonald / mapr-sparkml-streaming-uber Goto Github PK

scala> valuesDStream.foreachRDD {
     |  rdd => if (!rdd.isEmpty) {
     |  case class Uber(dt: String, lat: Double, lon: Double, base: String) extends Serializable
     |  //def parseUber(str: String): Uber = Uber(str(0), str(1).toDouble, str(2).toDouble, str(3))
     |  val count = rdd.count
     |   println("count received " + count)
     |  val sqlContext= new org.apache.spark.sql.SQLContext(sc)
     |  import sqlContext.implicits._
     |  rdd.collect().foreach(println)
     |  val df = rdd.map(_.split(",")).map(e => Uber(e(0), e(1).toDouble, e(2).toDouble, e(3)))
     |  val df1= sqlContext.createDataFrame(df,schema)
     |  df1.show()
     |  }
     |  }
<console>:72: error: overloaded method value createDataFrame with alternatives:
  (data: java.util.List[_],beanClass: Class[_])org.apache.spark.sql.DataFrame <and>
  (rdd: org.apache.spark.api.java.JavaRDD[_],beanClass: Class[_])org.apache.spark.sql.DataFrame <and>
  (rdd: org.apache.spark.rdd.RDD[_],beanClass: Class[_])org.apache.spark.sql.DataFrame <and>
  (rows: java.util.List[org.apache.spark.sql.Row],schema: org.apache.spark.sql.types.StructType)org.apache.spark.sql.DataFrame <and>
  (rowRDD: org.apache.spark.api.java.JavaRDD[org.apache.spark.sql.Row],schema: org.apache.spark.sql.types.StructType)org.apache.spark.sql.DataFrame <and>
  (rowRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row],schema: org.apache.spark.sql.types.StructType)org.apache.spark.sql.DataFrame
 cannot be applied to (org.apache.spark.rdd.RDD[Uber], org.apache.spark.sql.types.StructType)
        val df1= sqlContext.createDataFrame(df,schema)

`val df = rdd.map(_.split(",")).map(e => Uber(e(0), e(1).toDouble, e(2).toDouble, e(3)))
val df1 = df.toDF()`