在数据量方面,火花流的限制是什么？

Question

我有数千万行数据.是否可以使用火花流在一周或一天内分析所有这些？在数据量方面,火花流的限制是什么？我不确定什么是上限,什么时候我应该将它们放入我的数据库,因为Stream可能无法再处理它们了.我也有不同的时间窗口1,3,6小时等,我使用窗口操作来分隔数据.

请在下面找到我的代码:

conf = SparkConf().setAppName(appname)
sc = SparkContext(conf=conf)
ssc = StreamingContext(sc,300)
sqlContext = SQLContext(sc)
channels = sc.cassandraTable("abc","channels")
topic = 'abc.crawled_articles'
kafkaParams = {"metadata.broker.list": "0.0.0.0:9092"}

category = 'abc.crawled_article'
category_stream = KafkaUtils.createDirectStream(ssc, [category], kafkaParams)
category_join_stream = category_stream.map(lambda x:read_json(x[1])).filter(lambda x:x!=0).map(lambda x:categoryTransform(x)).filter(lambda x:x!=0).map(lambda x:(x['id'],x))


article_stream = KafkaUtils.createDirectStream(ssc, [topic], kafkaParams)
article_join_stream=article_stream.map(lambda x:read_json(x[1])).filter(lambda x: x!=0).map(lambda x:TransformInData(x)).filter(lambda x: x!=0).flatMap(lambda x:(a for a in x)).map(lambda x:(x['id'].encode("utf-8") ,x))

#axes topic  integration the article and the axes
axes_topic = 'abc.crawled_axes'
axes_stream = KafkaUtils.createDirectStream(ssc, [axes_topic], kafkaParams)
axes_join_stream = axes_stream.filter(lambda x:'delete' not in str(x)).map(lambda x:read_json(x[1])).filter(lambda x: x!=0).map(lambda x:axesTransformData(x)).filter(lambda x: x!=0).map(lambda x:(str(x['id']),x)).map(lambda x:(x[0],{'id':x[0], 'attitudes':x[1]['likes'],'reposts':0,'comments':x[1]['comments'],'speed':x[1]['comments']}))
#axes_join_stream.reduceByKeyAndWindow(lambda x, y: x + y, 30, 10).transform(axestrans).pprint()

#join
statistics = article_join_stream.window(1*60*60,5*60).cogroup(category_join_stream.window(1*60*60,60)).cogroup((axes_join_stream.window(24*60*60,5*60)))
statistics.transform(joinstream).pprint()

ssc.start()    # Start the computation ssc.awaitTermination()
ssc.awaitTermination()

Answer 1

一次一个：

• 是否可以在[给定的时间内]分析[一些大量的行]？

一般来说，是的 - Spark 允许您跨多台机器进行横向扩展，因此原则上您应该能够启动大型集群并在相对较短的时间内处理大量数据（假设我们谈论的是数小时或数天，而不是数秒或更短时间，由于开销，这可能会出现问题）。

具体来说，在我看来，在合理的时间内（即不使用非常大的集群）对数千万条记录执行您的问题中所示的处理是可行的。

• Spark Streaming 在数据量方面有什么限制？

我不知道，但你会很难做到这一点。有一些非常大的部署示例，例如在eBay中（“平均每天超过 30TB 的数百个指标”）。另请参阅常见问题解答，其中提到了一个由 8000 台机器组成的集群并处理 PB 级的数据。

• 什么时候应该将结果写入[某种存储]？

根据Spark-Streaming的基本模型，数据以微批次的方式进行处理。如果你的数据确实是一个流（即没有明确的结尾），那么最简单的方法就是存储每个RDD（即微批次）的处理结果。

如果您的数据不是流，例如您时不时地处理一堆静态文件，您可能应该考虑放弃流部分（例如仅使用 Spark 作为批处理器）。

由于您的问题提到了几个小时的窗口大小，我怀疑您可能需要考虑批处理选项。

• 如何在不同的时间窗口处理相同的数据？

如果您使用 Spark-Streaming，您可以维护多个状态（例如使用mapWithState） - 每个时间窗口一个状态。

另一个想法（代码更简单，操作更复杂） - 您可以启动多个集群，每个集群都有自己的窗口，从同一个流读取。

如果您正在进行批处理，则可以使用不同的时间窗口（例如reduceByWindow使用多个窗口大小）多次运行相同的操作。