小编Yin*_*Tan的帖子

我正在编写一个需要在特定HDFS路径中获取新二进制文件的组件,以便我可以根据这些数据进行一些在线学习.所以,我想在流中读取Flume从HDFS创建的二进制文件.我找到了spark API提供的几个函数,比如

public JavaDStream<byte[]> binaryRecordsStream(String directory,int recordLength)

和

public <K,V,F extends org.apache.hadoop.mapreduce.InputFormat<K,V>> 
     JavaPairInputDStream<K,V> fileStream(String directory, Class<K> kClass, Class<V> vClass, Class<F> fClass)

但是,我真的不知道如何使用这些功能.我试过binaryRecordStream,但它定义了文件的具体长度,所以它并不好.

对于fileStream功能,我用过:

SparkConf sparkConf = new   SparkConf().setAppName("SparkFileStreamTest").setMaster("local[2]");

// Create the context with the specified batch size
JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(durationInMillis));

JavaPairInputDStream<LongWritable, BytesWritable> inputDStream = jssc.fileStream(hdfsPath, LongWritable.class, BytesWritable.class, CustomInputFormat.class);

//**********************************************************************
JavaPairInputDStream<LongWritable, BytesWritable> inputDStream = jssc.fileStream(
            hdfsPath, LongWritable.class, BytesWritable.class, CustomInputFormat.class);

JavaDStream<byte[]> content = inputDStream.map(new Function<Tuple2<LongWritable, BytesWritable>, byte[]>() {
    @Override
    public byte[] call(Tuple2<LongWritable, BytesWritable> tuple2) …