title: spark streaming fileStream
body: <p>I'm programming with spark streaming but have some trouble with scala. I'm trying to use the function StreamingContext.fileStream</p>
  
  <p>The definition of this function is like this:</p>
  
  <pre><code>def fileStream[K, V, F &lt;: InputFormat[K, V]](directory: String)(implicit arg0: ClassManifest[K], arg1: ClassManifest[V], arg2: ClassManifest[F]): DStream[(K, V)]
  </code></pre>
  
  <p>Create a input stream that monitors a Hadoop-compatible filesystem for new files and reads them using the given key-value types and input format. File names starting with . are ignored.
  K
  Key type for reading HDFS file
  V
  Value type for reading HDFS file
  F
  Input format for reading HDFS file
  directory
  HDFS directory to monitor for new file</p>
  
  <p>I don't know how to pass the type of Key and Value. 
  My Code in spark streaming:</p>
  
  <pre><code>val ssc = new StreamingContext(args(0), "StreamingReceiver", Seconds(1),
    System.getenv("SPARK_HOME"), Seq("/home/mesos/StreamingReceiver.jar"))
  
  // Create a NetworkInputDStream on target ip:port and count the
  val lines = ssc.fileStream("/home/sequenceFile")
  </code></pre>
  
  <p>Java code to write the hadoop file:</p>
  
  <pre><code>public class MyDriver {
  
  private static final String[] DATA = { "One, two, buckle my shoe",
          "Three, four, shut the door", "Five, six, pick up sticks",
          "Seven, eight, lay them straight", "Nine, ten, a big fat hen" };
  
  public static void main(String[] args) throws IOException {
      String uri = args[0];
      Configuration conf = new Configuration();
      FileSystem fs = FileSystem.get(URI.create(uri), conf);
      Path path = new Path(uri);
      IntWritable key = new IntWritable();
      Text value = new Text();
      SequenceFile.Writer writer = null;
      try {
          writer = SequenceFile.createWriter(fs, conf, path, key.getClass(),
                  value.getClass());
          for (int i = 0; i &lt; 100; i++) {
              key.set(100 - i);
              value.set(DATA[i % DATA.length]);
              System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key,
                      value);
              writer.append(key, value);
          }
      } finally {
          IOUtils.closeStream(writer);
      }
  }
  </code></pre>
  
  <p>}</p>