我有一个csv文件;我在pyspark中将其转换为DataFrame(df);经过一些转换之后;我想在df中添加一个列;该列应该是简单的行id(从0或1开始到N)。
我在rdd中转换了df并使用“zipwithindex”。我把得到的rdd转换回df。这种方法可以工作,但是它生成了25000个任务,并且需要大量的执行时间。我想知道是否有其他方法可以减少运行时间。
下面是我的代码片段;我正在处理的csv文件很大;包含数十亿行。
debug_csv_rdd = (sc.textFile("debug.csv")
.filter(lambda x: x.find('header') == -1)
.map(lambda x : x.replace("NULL","0")).map(lambda p: p.split(','))
.map(lambda x:Row(c1=int(x[0]),c2=int(x[1]),c3=int(x[2]),c4=int(x[3]))))
debug_csv_df = sqlContext.createDataFrame(debug_csv_rdd)
debug_csv_df.registerTempTable("debug_csv_table")
sqlContext.cacheTable("debug_csv_table")
r0 = sqlContext.sql("SELECT c2 FROM debug_csv_table WHERE c1 = 'str'")
r0.registerTempTable("r0_table")
r0_1 = (r0.flatMap(lambda x:x)
.zipWithIndex()
.map(lambda x: Row(c1=x[0],id=int(x[1]))))
r0_df=sqlContext.createDataFrame(r0_2)
r0_df.show(10)
也可以使用sql包中的函数。它将生成一个唯一的id,但是它不是连续的,因为它取决于分区的数量。我相信Spark 1.5+
编辑:19/1/2017
由@Sean评论
使用
monotonically_increasing_id()
代替Spark 1.6和on相关问题 更多 >
编程相关推荐