【大数据开发】SparkStreaming基于HDFS的实时计算开发（四）

Part

StreamingContext.socketTextStream (）

streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory)
streamingContext.textFileStream[KeyClass, ValueClass, InputFormatClass](dataDirecto

关于集群同步的时间问题：

修改主机时间，使源数据节点时间必须与计算节点时间保持同步：

修改hadoop配置文件：

本地创建源数据文件：

查看文件：

创建输入流数据源目录：

//Java
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;


import java.util.Arrays;


public class HDFSWordCountJava {
public static void main(String[] args) throws Exception{
SparkConf cOnf= new SparkConf().setMaster("local[1]").setAppName("HDFSworkWordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(3));  //每隔3秒处理数据
String filePath = "hdfs://bigdata-pro-m01.kfk.com:9000/user/spark/datas/sparkstreaming";


JavaDStream<String> lines = jssc.textFileStream(filePath);


JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());


JavaPairDStream<String,Integer> pair = words.mapToPair(word -> new Tuple2<>(word,1));


JavaPairDStream<String,Integer> wordcount = pair.reduceByKey((x,y) -> (x + y));


        wordcount.print();
jssc.start();
jssc.awaitTermination();
}
}

//Scala
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Durations, StreamingContext}


object HDFSWordCountScala {
def main(args: Array[String]): Unit = {
val cOnf= new SparkConf().setMaster("local[1]").setAppName("HDFSworkWordCount")


val jssc = new StreamingContext(conf, Durations.seconds(3))


val filePath: String = "hdfs://bigdata-pro-m01.kfk.com:9000/user/spark/datas/sparkstreaming" //sparkstreaming目录下新增的所有文件都会被处理


val lines = jssc.textFileStream(filePath)


val words = lines.flatMap(line => line.split(" "));


val pair = words.map(word => (word,1))


val wordcount = pair.reduceByKey((x , y) => x + y)


wordcount.print()
jssc.start()
jssc.awaitTermination()
}
}

1. 启动程序

   
   

2. 上传HDFS文件
   

    这里为什么上传了这么多文件没有执行成功呢？这是因为开始没有同步集群时间导致Last Modified异常（20200114测试的），计算节点无法实时读取数据。最后修改时间后重启namenode和datanode服务，重新上传6.txt文件才被计算节点处理。

    3. 打印结果

    总结一下，在实际企业中HDFS用于实时数据的传输比较少，这是因为数据落地到HDFS分布式文件系统需要一段时间，从而影响了时效性，所以分布式存储文件多用于大数据离线处理。而数据实时性处理还是要Kafka这种响应快速的中间件来映射。