Spark sql Dataframe - import sqlContext.implicits._

Question

mam główny, który tworzy kontekst zapłonową:

val sc = new SparkContext(sparkConf) 
    val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
    import sqlContext.implicits._ 

Następnie tworzy dataframe i robi filtry i walidacje na dataframe.

val convertToHourly = udf((time: String) => time.substring(0, time.indexOf(':')) + ":00:00") 

    val df = sqlContext.read.schema(struct).format("com.databricks.spark.csv").load(args(0)) 
    // record length cannot be < 2 
    .na.drop(3) 
    // round to hours 
    .withColumn("time",convertToHourly($"time")) 

Działa to świetnie.

ale gdy próbuję porusza moje walidacji do innego pliku, wysyłając dataframe do

function ValidateAndTransform(df: DataFrame) : DataFrame = {...} 

że dostaje Dataframe & robi walidacji i przekształceń: Wydaje się, że muszę

import sqlContext.implicits._ 

To avoid the error: “value $ is not a member of StringContext” that happens on line: .withColumn("time",convertToHourly($"time"))

Ale do korzystania z import sqlContext.implicits._ potrzebuję również sqlContext zdefiniowanego w nowym złożyć tak:

val sc = new SparkContext(sparkConf) 
val sqlContext = new org.apache.spark.sql.SQLContext(sc) 

lub wysłać go do

function ValidateAndTransform(df: DataFrame) : DataFrame = {...} 
function 

czuję się separacji próbuję zrobić do 2 plików (główny & walidacji) nie jest wykonane prawidłowo ...

Każdy pomysł, jak to zaprojektować? Lub po prostu wyślij sqlContext do funkcji?

Dzięki!

Answer 1

Można pracować z pojedynczą instancją klasy SQLContext. Można spojrzeć na ten przykład w spark repository

/** Lazily instantiated singleton instance of SQLContext */ 
object SQLContextSingleton { 

    @transient private var instance: SQLContext = _ 

    def getInstance(sparkContext: SparkContext): SQLContext = { 
    if (instance == null) { 
     instance = new SQLContext(sparkContext) 
    } 
    instance 
    } 
} 
... 
//And wherever you want you can do 
val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext) 
import sqlContext.implicits._