스칼라에서 스칼라를 사용하여 DataSet을 사용하는 방법은 무엇입니까?

Question

나는 그때 DataSet에 변환을 내 CSV 사용 DataFrame를로드하지만,이 라인에서이

여러 마커와 같은 쇼의 :
- 데이터 집합에 저장 유형에 대한 인코더를 찾을 수 없습니다.
spark.implicits._ 다른 유형을 직렬화하기위한 지원이 향후 릴리스에서 추가 될 예정입니다.
- 메서드에 대한 인수가 충분하지 않습니다. (암시 적 증거 $ 2 :
org.apache.spark.sql.Encoder) [org.apache.spark.sql.Dataset] [DataSet.spark.sql.Dataset] . 알 수없는 값 매개 변수 증명 $ 2

해결 방법?.

case class aaCSV(
    a: String, 
    b: String 
    ) 

object WorkShop { 

    def main(args: Array[String]) = { 
    val conf = new SparkConf() 
     .setAppName("readCSV") 
     .setMaster("local") 
    val sc = new SparkContext(conf) 
    val sqlContext = new SQLContext(sc) 
    val customSchema = StructType(Array(
     StructField("a", StringType, true), 
     StructField("b", StringType, true))) 

    val df = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").schema(customSchema).load("/xx/vv/ss.csv") 
    df.printSchema() 
    df.show() 
    val googleDS = df.as[aaCSV] 
    googleDS.show() 

    } 

} 

지금 나는이 같은 주요 기능을 변경 - - 내 코드는

def main(args: Array[String]) = { 
    val conf = new SparkConf() 
     .setAppName("readCSV") 
     .setMaster("local") 
    val sc = new SparkContext(conf) 
    val sqlContext = new SQLContext(sc) 
import sqlContext.implicits._; 
    val sa = sqlContext.read.csv("/xx/vv/ss.csv").as[aaCSV] 
    sa.printSchema() 
    sa.show() 
} 

를하지만 오류가 발생합니다 - "주"org.apache.spark.sql.AnalysisException 스레드 예외 : 할 수 없습니다 주어진 입력 열을 'Adj_Close'해결 : [_c1, _c2, _c5, _c4, _c6, _c3, _c0]; line 1 pos 7. 어떻게해야합니까?

이제 스파크 스케줄러를 사용하여 주어진 시간 간격을 기준으로 내 방법을 실행합니다. 그러나이 링크 (https://spark.apache.org/docs/latest/job-scheduling.html#scheduling-within-an-application)를 참조합니다. 친절하게 도와주세요.

Answer 1

https://spark.apache.org/docs/latest/sql-programming-guide.html#creating-datasets 당신이 당신의 CSV 파일의 헤더 (열 이름)이 있습니까? 예인 경우 read 문에 .option("header","true")을 추가하십시오. 예 : sqlContext.read.option("header","true").csv("/xx/vv/ss.csv").as[aaCSV]. http://technippet.blogspot.in/2016/10/different-ways-of-creating.html

Answer 2

DF을 DS으로 변환하기 전에 아래의 가져 오기를 추가해보십시오.

sc.implicits._ 

또는 데이터 집합 작업에 대한 추가 정보를 원하시면

sqlContext.implicits._