相关疑难解决方法(0)

RDD具有一个有意义的(与存储模型强加的一些随机顺序相反),如果它被处理sortBy(),则如本回复中所解释的那样.

现在,哪些操作保留了该订单？

例如,是否保证(之后a.sortBy())

a.map(f).zip(a) === 
a.map(x => (f(x),x))

怎么样

a.filter(f).map(g) === 
a.map(x => (x,g(x))).filter(f(_._1)).map(_._2)

关于什么

a.filter(f).flatMap(g) === 
a.flatMap(x => g(x).map((x,_))).filter(f(_._1)).map(_._2)

这里"平等" ===被理解为"功能等同",即,没有办法使用用户级操作来区分结果(即,没有读取日志和c).

我正在使用HDFS在集群上使用Apache Spark.据我了解,HDFS正在数据节点上分发文件.因此,如果在文件系统上放置"file.txt",它将被拆分为分区.现在我在打电话

rdd = SparkContext().textFile("hdfs://.../file.txt") 

来自Apache Spark.rdd现在自动与文件系统上的"file.txt"分区相同吗？我打电话时会发生什么

rdd.repartition(x)

其中x>那么hdfs使用的分区？Spark会在物理上重新排列hdfs上的数据以在本地工作吗？

示例:我在HDFS系统上放置了一个30GB的文本文件,它将它分发到10个节点上.Will Spark a)使用相同的10个分区吗？和b)当我调用重新分区(1000)时,在群集中洗牌30GB？

我试图解决向数据集添加序列号的古老问题.我正在使用DataFrames,似乎没有相应的DataFrame RDD.zipWithIndex.另一方面,以下工作或多或少按我希望的方式工作:

val origDF = sqlContext.load(...)    

val seqDF= sqlContext.createDataFrame(
    origDF.rdd.zipWithIndex.map(ln => Row.fromSeq(Seq(ln._2) ++ ln._1.toSeq)),
    StructType(Array(StructField("seq", LongType, false)) ++ origDF.schema.fields)
)

在我的实际应用程序中,origDF不会直接从文件中加载 - 它将通过将2-3个其他DataFrame连接在一起而创建,并将包含超过1亿行.

有一个更好的方法吗？我该怎么做才能优化它？