想象一下我有一个Dask来自read_csv或以其他方式创建的 DataFrame。
如何为 dask 数据框创建唯一索引？
笔记：
reset_index在每个分区中构建单调升序索引。这意味着分区 1 为 (0,1,2,3,4,5,... )，分区 2 为 (0,1,2,3,4,5,... )，(0,1,2 ,3,4,5,... ) 对于分区 3 等等。
我想要数据帧中的每一行（跨所有分区）都有一个唯一的索引。
我有DocsRDD:RDD [String,String]
val DocsRDD = sc.wholeTextFiles("myDirectory/*" , 2)
DocsRDD:
Doc1.txt , bla bla bla .....\n bla bla bla \n bla ... bla
Doc2.txt , bla bla bla .....bla \n bla bla \n bla ... bla
Doc3.txt , bla bla bla .....\n bla bla bla \n bla ... bla
Doc4.txt , bla bla \n .....\n bla bla bla bla \n ... bla
有没有一种高效,优雅的方法从mapPartitions中提取n-gram？到目前为止,我已经尝试了所有的东西,我已经阅读了至少5次关于mapPartitions的一切,但我仍然无法理解如何使用它!似乎太难以操纵了.总之我想:
val NGramsRDD = DocsRDD.map(x => (x._1 , x._2.sliding(n) ) )
但是有效地使用mapPartitions.我对mapPartitions的基本误解是:
OneDocRDD:RDD [String]
val OneDocRDD = sc.textFile("myDoc1.txt" , …我试图了解Scala中的类型参数.我们来看下面的一般例子:
def func1[T](a : T) : T = a
我知道func1接受任何类型的1个参数,并返回相同类型的参数.我不明白的原因是:
def func1[T]
为什么[T]在function1之后？在func1之后我们可以简单地在没有[T]的情况下编写它,例如:
def func1(a : T) : T = a
1)在func1之后,[T]意味着什么？为什么我们把它放在那里？
2)为什么我们对课程做同样的事情？
class MyClass[T]{...}
我的意思是MyClass实例化是MyClass类型.那里[T]意味着什么？你不是说我有一个MyClass类型的布尔类,你说我有一个类型为MyClass的对象吗？
提前致谢.
想象一下我有一个三胞胎：
val RecordRDD : RDD[Int, String, Int] = {
(5 , "x1", 100),
(3 , "x2", 200),
(3 , "x4", 300),
(5 , "x1", 150),
(3 , "x2", 160),
(5 , "x1", 400)
}
如何有效地按前两个元素对它们进行分组并按第三个元素排序？例如制作：
[5 , [ "x1" -> [100, 150, 400] ]
[3 , [ ["x2" -> [160, 200]], ["x4" -> [300]] ]
我正在寻找一种有效的方法。
我应该将其设为 DataFrame 并使用 GroupBy(Col1,Col2) 和 SortBy(Col3) 吗？
这会比 Spark RDD 的 groupBy 更有效吗？
AggregateByKey 可以同时聚合 2 个键吗？
*你可以假设这个RDD相当大！提前致谢。
看来我处于死胡同!好吧这是一个我几乎已经完成的编程任务但是我已经在一个巨大的问题中停留了几天!
def filterAcc(p: Tweet => Boolean, acc: TweetSet): TweetSet = {
if(p(elem))
acc.incl(elem)
if(!(left isEmpty))
left filterAcc(p, acc)
if(!(right isEmpty))
right filterAcc(p, acc)
acc
}
经过多次println和调试后,似乎acc.incl(elem)似乎不起作用!我不知道为什么,但它没有添加元素.我称之为:
def filter(p: Tweet => Boolean): TweetSet = filterAcc(p, new Empty)
这也是来自Scala中的编程分配:面向对象的集合.任何帮助,将不胜感激.谢谢.