代码之家 › 专栏 › 技术社区 › NamrataK

.withColumn不提供dataframe的原始列,只提供新添加的列

apache-spark-sql apache-spark

NamrataK · 技术社区 · 6 年前

val withOneDayts=userDataFrame.join(articleDataFrame,userDataFrame("cid")===articleDataFrame("id"),"left").drop(articleDataFrame("id")).drop(articleDataFrame("published_at"))
val frame = withOneDayts.withColumn("views", getViews(withOneDayts("timestamp"), withOneDayts("oneDay_ts"))).groupBy("cid").agg(sum("views"))
val with15MinViews = withOneDayts.withColumn("views", getViews(withOneDayts("timestamp"), withOneDayts("15_min_ts"))).groupBy("cid").agg(sum("views")).withColumnRenamed("sum(views)","views_in_15_min").drop("15_min_ts")
val with30MinViews = with15MinViews.withColumn("views",getViews(with15MinViews("timestamp"),with15MinViews("30_min_ts"))).groupBy("cid").agg(sum("views")).withColumnRenamed("sum(views)","views_in_30_min").drop("30_min_ts")

我试图在文章发表后的前15分钟和30分钟内获得观点。但对于30分钟的视图,它给出了错误“无法解析[cid,views\u in\u 15\u min]之间的“timestamp”

1 回复 | 直到 6 年前

koiralo 6 年前

错误出现在使用数据帧的第四行 with15MinViews 其中不包含字段 timestamp

val with30MinViews = with15MinViews.withColumn("views",
      getViews(with15MinViews("timestamp"),with15MinViews("30_min_ts"))
    )  
    .groupBy("cid").agg(sum("views"))
    .withColumnRenamed("sum(views)","views_in_30_min")
.drop("30_min_ts")

这个 有15条新闻 dataframe仅包含在上使用的列 groypBy 和 aggregation .所以它只包含 cid ,则, views_in_15_min

我希望这有帮助!

推荐文章

srinath tripuraneni · {DataFrameWriter CSV到HDFS文件系统}不分区写入数据

2 年前

Calcutta · Google Colab中的Spark SQL在大数据上失败

2 年前

Palkin Jangra · 使用循环在Pyspark数组元素上和元素本身内迭代两次

2 年前

Doraemon · PySpark:使用不同值的字符串类型列创建聚合列

2 年前

amol visave · spark作业失败时会发生什么?

2 年前

Alex Jolly · 如何在另一个pyspark数据帧中查询开始时间和结束时间之间的日期时间

2 年前

chun · pyspark dataframe在s3中两次写入csv文件

2 年前

Mod_x · 如何将特定列的行标题更改为行标题,并在pySpark中生成矩阵?

2 年前

katty · 动态地将参数传递给scala中的函数

6 年前

Abhishek Choudhary · 如何更新数组列?

6 年前