火花流FILESTREAM(spark streaming fileStream)

我与编程火花流，但有一些麻烦与斯卡拉。 我想使用的功能StreamingContext.fileStream

这个函数的定义是这样的：

def fileStream[K, V, F <: InputFormat[K, V]](directory: String)(implicit arg0: ClassManifest[K], arg1: ClassManifest[V], arg2: ClassManifest[F]): DStream[(K, V)]

创建监视新文件在Hadoop兼容的文件系统，并使用特定键值的类型和输入格式读取它们一个输入流。 文件名开头。 被忽略。 K键式读取HDFS文件V值类型读取HDFS文件F输入格式读取HDFS文件目录HDFS目录监视新文件

我不知道如何通过键和值的类型。 我在火花流代码：

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")

Java代码编写Hadoop的文件：

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 < 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);
    }
}

}

Answer 1:

如果你想使用fileStream ，你将不得不调用时，它提供所有类型3个PARAMS它。 你需要知道你的Key ， Value和InputFormat类型调用它。 如果你的类型是LongWritable ， Text和TextInputFormat ，你会打电话fileStream像这样：

val lines = ssc.fileStream[LongWritable, Text, TextInputFormat]("/home/sequenceFile")

如果这些3种类型碰巧是你的类型，那么你可能想使用textFileStream ，而不是因为它不需要任何类型PARAMS和代表们fileStream使用这些3种类型，我提到。 使用是这样的：

val lines = ssc.textFileStream("/home/sequenceFile")

Answer 2:

val filterF = new Function[Path, Boolean] {
    def apply(x: Path): Boolean = {
      val flag = if(x.toString.split("/").last.split("_").last.toLong < System.currentTimeMillis) true else false
      return flag
    }
}

val streamed_rdd = ssc.fileStream[LongWritable, Text, TextInputFormat]("/user/hdpprod/temp/spark_streaming_input",filterF,false).map(_._2.toString).map(u => u.split('\t'))

文章来源: spark streaming fileStream