我最近用sparksql对一个静态日志文件进行了分析（找出像ip地址这样出现超过10次的内容）。问题来自此站点。但是我用了我自己的实现。我将日志读入rdd，将rdd转换为Dataframe（在pojo的帮助下），并使用Dataframe操作。
现在我要用spark streaming做一个类似的分析，对一个流日志文件进行30分钟的分析，以及一天的汇总结果。在这里可以再次找到解决方案，但我想用另一种方法。所以我所做的就是
使用flume将数据从日志文件写入hdfs目录
使用javadstream从hdfs读取.txt文件
那我就不知道该怎么办了。这是我使用的代码
Long slide = 10000L; //new batch every 10 seconds
Long window = 1800000L; //30 mins
SparkConf conf = new SparkConf().setAppName("StreamLogAnalyzer");
JavaStreamingContext streamingContext = new JavaStreamingContext(conf, new Duration(slide));
JavaDStream<String> dStream = streamingContext.textFileStream(hdfsPath).window(new Duration(window), new Duration(slide));现在我似乎无法决定是否应该将每个批处理都转换为一个Dataframe，并像以前那样处理静态日志文件。或者说这种方式既费时又费事。
我是一个绝对的noob流以及Flume。有人能给我指点一下吗？
1条答案
按热度按时间d7v8vwbk1#
在spark的最新版本中，在spark中使用dataframe（和dataset）是最受欢迎的方式，因此使用它是一个正确的选择。我认为，当您将文件移动到hdfs而不是从任何事件日志中读取时，由于流的非显式性质，会出现一些模糊性。
这里的要点是选择正确的批处理时间大小（或片段中的幻灯片大小），这样应用程序将处理在该时间段下加载的数据，而不会出现批处理队列。