首页 / PYTHON / python – 如何在pySpark数据帧中添加行ID [复制]

python – 如何在pySpark数据帧中添加行ID [复制]

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了python – 如何在pySpark数据帧中添加行ID [复制]，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含1404字，纯文字阅读大概需要3分钟。

内容图文

参见英文答案 > Primary keys with Apache Spark                                    3个
我有一个csv文件;我在pyspark中转换为DataFrame(df);经过一番改造;我想在df中添加一列;这应该是简单的行id(从0或1开始到N).

我在rdd中转换了df并使用“zipwithindex”.我将生成的rdd转换回df.这种方法有效,但它产生了250k的任务,并且需要花费大量的时间来执行.我想知道是否还有其他方法可以减少运行时间.

以下是我的代码片段;我正在处理的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中使用

from pyspark.sql.functions import monotonicallyIncreasingId

# This will return a new DF with all the columns + id
res = df.withColumn("id", monotonicallyIncreasingId())

编辑：19/1/2017

由@Sean评论

使用monotonically_increasing_id()代替Spark 1.6和on

内容总结

以上是互联网集市为您收集整理的python – 如何在pySpark数据帧中添加行ID [复制]全部内容，希望文章能够帮你解决python – 如何在pySpark数据帧中添加行ID [复制]所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/709949.html

来源：【匿名】