Spark sql Dataframe - nhập khẩu sqlContext.implicits._

Question

tôi có chính mà tạo ra bối cảnh spark:

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

Sau đó tạo dataframe và làm bộ lọc và kiểm chứng thực trên 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")) 

Điều này thật tuyệt vời.

NHƯNG Khi tôi cố gắng di chuyển kiểm chứng thực của tôi vào tập tin khác bằng cách gửi dataframe để

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

mà được các Dataframe & không kiểm chứng thực và biến đổi: Nó có vẻ như tôi cần

import sqlContext.implicits._ 

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

Nhưng để sử dụng import sqlContext.implicits._ Tôi cũng cần sqlContext hoặc được xác định trong nộp như sau:

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

hoặc gửi nó vào

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

Tôi cảm thấy như tách Tôi đang cố gắng làm gì để 2 file (chính & xác nhận) không được thực hiện một cách chính xác ...

Bất kỳ ý tưởng nào về cách thiết kế này? Hoặc đơn giản là gửi sqlContext đến hàm?

Cảm ơn!

Answer 1

Bạn có thể làm việc với một phiên bản đơn lẻ của SQLContext. Bạn có thể có một cái nhìn vào ví dụ này trong 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._