<commit_before>package org.apache.spark.sql.streaming

import org.apache.spark.Logging
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.sources.{DeletableRelation, DestroyRelation}
import org.apache.spark.sql.types.StructType
import org.apache.spark.streaming.dstream.DStream

/**
 * Created by ymahajan on 25/09/15.
 */
case class FileStreamRelation(@transient val sqlContext: SQLContext,
                              options: Map[String, String],
                              override val schema: StructType)
  extends StreamBaseRelation with DeletableRelation
  with DestroyRelation with Logging with StreamPlan with Serializable {

  val DIRECTORY = "directory"
  // HDFS directory to monitor for new file
  val KEY = "key:"
  // Key type for reading HDFS file
  val VALUE = "value"
  //Value type for reading HDFS file
  val INPUT_FORMAT_HDFS = "inputformathdfs" //Input format for reading HDFS file

  val FILTER = "filter"
  //Function to filter paths to process
  val NEW_FILES_ONLY = "newfilesonly"
  //Should process only new files and ignore existing files in the directory
  val CONF = "conf" //Hadoop configuration

  val directory = options(DIRECTORY)

  val context = StreamingCtxtHolder.streamingContext

  val fileStream: DStream[String] = context.textFileStream(directory)
  //TODO: Yogesh, add support for other types of files streams

  private val streamToRow = {
    try {
      val clz = StreamUtils.loadClass(options("streamToRow"))
      clz.newInstance().asInstanceOf[MessageToRowConverter]
    } catch {
      case e: Exception => sys.error(s"Failed to load class : ${e.toString}")
    }
  }

  @transient val stream: DStream[InternalRow] = fileStream.map(streamToRow.toRow)

  override def destroy(ifExists: Boolean): Unit = {
    throw new IllegalAccessException("Stream tables cannot be dropped")
  }

  override def delete(filterExpr: String): Int = {
    throw new IllegalAccessException("Stream tables cannot be dropped")
  }

  def truncate(): Unit = {
    throw new IllegalAccessException("Stream tables cannot be truncated")
  }
}<commit_msg>Making streamingContext transient to avoid serialization issue<commit_after>package org.apache.spark.sql.streaming

import org.apache.spark.Logging
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.sources.{DeletableRelation, DestroyRelation}
import org.apache.spark.sql.types.StructType
import org.apache.spark.streaming.dstream.DStream

/**
 * Created by ymahajan on 25/09/15.
 */
case class FileStreamRelation(@transient val sqlContext: SQLContext,
                              options: Map[String, String],
                              override val schema: StructType)
  extends StreamBaseRelation with DeletableRelation
  with DestroyRelation with Logging with StreamPlan with Serializable {

  val DIRECTORY = "directory"
  // HDFS directory to monitor for new file
  val KEY = "key:"
  // Key type for reading HDFS file
  val VALUE = "value"
  //Value type for reading HDFS file
  val INPUT_FORMAT_HDFS = "inputformathdfs" //Input format for reading HDFS file

  val FILTER = "filter"
  //Function to filter paths to process
  val NEW_FILES_ONLY = "newfilesonly"
  //Should process only new files and ignore existing files in the directory
  val CONF = "conf" //Hadoop configuration

  val directory = options(DIRECTORY)

  @transient val context = StreamingCtxtHolder.streamingContext

  @transient val fileStream: DStream[String] = context.textFileStream(directory)
  //TODO: Yogesh, add support for other types of files streams

  private val streamToRow = {
    try {
      val clz = StreamUtils.loadClass(options("streamToRow"))
      clz.newInstance().asInstanceOf[MessageToRowConverter]
    } catch {
      case e: Exception => sys.error(s"Failed to load class : ${e.toString}")
    }
  }

  @transient val stream: DStream[InternalRow] = fileStream.map(streamToRow.toRow)

  override def destroy(ifExists: Boolean): Unit = {
    throw new IllegalAccessException("Stream tables cannot be dropped")
  }

  override def delete(filterExpr: String): Int = {
    throw new IllegalAccessException("Stream tables cannot be dropped")
  }

  def truncate(): Unit = {
    throw new IllegalAccessException("Stream tables cannot be truncated")
  }
}
{'fraction_non_alphanumeric': 0.0856815578465063, 'fraction_numerical': 0.0027491408934707906, 'mean_word_length': 3.7611777535441657, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3876450', 'n_tokens_mistral': 1289, 'n_tokens_neox': 1223, 'n_words': 388}