title: Scala Spark Streaming files into a dataframe
body: <p>I am attempting to stream files from my local file system using Apache Spark in scala.</p>
  
  <p>I am running the following code in Eclipse with all the spark libraries (plus a couple of others):</p>
  
  <pre><code>package com.addlesee.sparkstreaming
  
  import org.apache.spark._
  import org.apache.spark.SparkContext._
  import org.apache.spark.streaming._
  import org.apache.spark.streaming.StreamingContext._
  import org.apache.log4j.Level
  import Utilities._
  import com.lucidworks.spark._
  import com.lucidworks.spark.SparkApp._
  import com.lucidworks.spark.SparkApp.StreamProcessor
  import org.apache.spark.sql._
  import org.apache.spark.sql.SparkSession._
  import org.apache.spark.sql.SQLContext._
  import org.apache.hadoop.fs._
  import org.apache.hadoop.fs.FileSystem._
  
  object fileRead {
    def main(args: Array[String]) {
  
      val conf = new SparkConf().setMaster("local[*]").setAppName("fileRead")
      val sc = new SparkContext(conf)
      val sqlContext = new SQLContext(sc)
      import sqlContext.implicits._
      val ssc = new StreamingContext(sc, Seconds(1))
  
      val options = Map("collection" -&gt; "filetester", "zkhost" -&gt; "localhost:9984", "gen_uniq_key" -&gt; "true")
      val lines = ssc.textFileStream("file:///home/angus/data/data/wikipedia-cats/copy2")
      print("lines= " + lines)
  
        lines.foreachRDD((rdd, time) =&gt; {
          if(rdd.count() &gt; 0){
            val rdd2 = rdd.repartition(1).cache()
            print("rdd= " + rdd)
            print("rdd2= " + rdd2)
            val df = rdd2.toDF()
  
            df.write.format("solr").options(options).mode(org.apache.spark.sql.SaveMode.Overwrite).save()
          }
      })
  
  
      ssc.start()
      ssc.awaitTermination()
    }
  }
  </code></pre>
  
  <p>and am recieving the following:</p>
  
  <pre><code>2017-08-25 16:27:52,425 [main] WARN  NativeCodeLoader  - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
  2017-08-25 16:27:52,546 [main] WARN  Utils  - Your hostname, angus-pc resolves to a loopback address: 127.0.1.1; using 192.118.36.174 instead (on interface wlp3s0)
  2017-08-25 16:27:52,546 [main] WARN  Utils  - Set SPARK_LOCAL_IP if you need to bind to another address
  2017-08-25 16:27:53,029 [main] INFO  log  - Logging initialized @1580ms
  2017-08-25 16:27:53,071 [main] INFO  Server  - jetty-9.3.z-SNAPSHOT
  2017-08-25 16:27:53,084 [main] INFO  Server  - Started @1635ms
  2017-08-25 16:27:53,095 [main] WARN  Utils  - Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
  2017-08-25 16:27:53,101 [main] INFO  AbstractConnector  - Started ServerConnector@13c43d25{HTTP/1.1,[http/1.1]}{0.0.0.0:4041}
  2017-08-25 16:27:53,126 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@71812481{/jobs,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,127 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@40238dd0{/jobs/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,128 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@79179359{/jobs/job,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,129 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@6a55299e{/jobs/job/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,130 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@4eb386df{/stages,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,131 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@3a0baae5{/stages/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,132 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@289710d9{/stages/stage,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,135 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@503ecb24{/stages/stage/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,136 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@6995bf68{/stages/pool,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,137 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@77825085{/stages/pool/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,138 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@71c27ee8{/storage,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,138 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@5b1ebf56{/storage/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,139 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@4b1d6571{/storage/rdd,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,140 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@3549bca9{/storage/rdd/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,144 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@6fb365ed{/environment,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,145 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@16414e40{/environment/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,146 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@525575{/executors,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,147 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@5a709816{/executors/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,148 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@6ad3381f{/executors/threadDump,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,148 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@db57326{/executors/threadDump/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,159 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@4748a0f9{/static,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,160 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@1941a8ff{/,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,165 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@149dd36b{/api,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,169 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@4f1bfe23{/jobs/job/kill,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,171 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@512535ff{/stages/stage/kill,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,459 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@56ace400{/metrics/json,null,AVAILABLE,@Spark}
  lines= org.apache.spark.streaming.dstream.MappedDStream@74a9c4b02017-08-25 16:27:53,852 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@1e34c607{/streaming,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,853 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@36b6964d{/streaming/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,854 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@aa5455e{/streaming/batch,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,854 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@5dda14d0{/streaming/batch/json,null,AVAILABLE,@Spark}
  2017-08-25 16:27:53,855 [main] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@3b4ef7{/static/streaming,null,AVAILABLE,@Spark}
  2017-08-25 16:28:46,307 [JobGenerator] INFO  FileInputFormat  - Total input paths to process : 1
  rdd= MapPartitionsRDD[106] at textFileStream at file2solr.scala:28rdd2= MapPartitionsRDD[110] at repartition at file2solr.scala:332017-08-25 16:28:47,564 [streaming-job-executor-0] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@30698a7f{/SQL,null,AVAILABLE,@Spark}
  2017-08-25 16:28:47,565 [streaming-job-executor-0] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@2ac04ffb{/SQL/json,null,AVAILABLE,@Spark}
  2017-08-25 16:28:47,565 [streaming-job-executor-0] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@1b62dd54{/SQL/execution,null,AVAILABLE,@Spark}
  2017-08-25 16:28:47,566 [streaming-job-executor-0] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@5b4a1f33{/SQL/execution/json,null,AVAILABLE,@Spark}
  2017-08-25 16:28:47,567 [streaming-job-executor-0] INFO  ContextHandler  - Started o.s.j.s.ServletContextHandler@4049fe04{/static/sql,null,AVAILABLE,@Spark}
  Exception in thread "streaming-job-executor-0" java.lang.NoClassDefFoundError: com/typesafe/scalalogging/LazyLogging
      at java.lang.ClassLoader.defineClass1(Native Method)
      at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
      at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
      at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
      at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
      at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
      at java.lang.Class.getDeclaredConstructors0(Native Method)
      at java.lang.Class.privateGetDeclaredConstructors(Class.java:2671)
      at java.lang.Class.getConstructor0(Class.java:3075)
      at java.lang.Class.newInstance(Class.java:412)
      at org.apache.spark.sql.execution.datasources.DataSource.write(DataSource.scala:470)
      at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:48)
      at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:58)
      at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:56)
      at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:74)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:117)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:117)
      at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:138)
      at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
      at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:135)
      at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:116)
      at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:92)
      at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:92)
      at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:610)
      at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:233)
      at com.addlesee.sparkstreaming.fileRead$$anonfun$main$1.apply(file2solr.scala:38)
      at com.addlesee.sparkstreaming.fileRead$$anonfun$main$1.apply(file2solr.scala:31)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
      at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:416)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
      at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
      at scala.util.Try$.apply(Try.scala:192)
      at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:257)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
      at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
      at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
      at java.lang.Thread.run(Thread.java:748)
  Caused by: java.lang.ClassNotFoundException: com.typesafe.scalalogging.LazyLogging
      at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:335)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
      ... 50 more
  </code></pre>
  
  <p>Firstly, I only receive my rdd prints and errors when I create a new file directly into the directory and then rename it... Moving, renaming or editing files that already exist do not output anything.</p>
  
  <p>I believe the problem is the <code>toDF()</code> function or my method to stream files.</p>
  
  <p>I am very new to Scala so must be misunderstanding something. I have printed lines, rdd and rdd2 with labels if that helps.</p>
  
  <p>I have tried other methods and running this in the spark-shell with no luck.</p>
  
  <p>Thank you for any help!</p>