Spark Streaming 的saveAsTextFiles遇到的坑

简介: 使用sparkStreaming消费数据,并使用Dstream的 saveAsTextFile保存数据到hdfs中,通过使用这个方法,生成的文件夹存在问题, 代码例子如下:     resultRdd.

使用sparkStreaming消费数据,并使用Dstream的 saveAsTextFile保存数据到hdfs中,通过使用这个方法,生成的文件夹存在问题,

代码例子如下:

    resultRdd.map(x=>x).saveAsTextFiles("hdfs:ip//data/storage/20181010/"+(new Date()))  //new Date()自行转化

    ssc.start()

    ssc.awaitermination()


而hsfs中目录显示为

   /data/storage/20181010/201810100708223-1547016648000

   /data/storage/20181010/201810100708223-1547016652000

   /data/storage/20181010/201810100708223-1547016658000

   .........................................


从中发现最后面多了一条横杠 -和时间戳1547016648000,是根据间隔时间自动生成的,但是我不想要他后面的-1547016648000,

并且201810100708223日期固定住了

查看saveAsTextFiles源码


def saveAsTextFiles(prefix: String, suffix: String = ""): Unit = ssc.withScope {

  val saveFunc = (rdd: RDD[T], time: Time) => {

    val file = rddToFileName(prefix, suffix,time)

    rdd.saveAsTextFile(file)

  }

  this.foreachRDD(saveFunc)

}

saveAsTextFiles方法中也是调用了saveAsTextFile方法,其中有个添加时间戳的方法。


于是我根据源码自己使用foreachRDD,生成文件使用saveAsTextFile


resultRdd.foreachRDD{

rdd=>{}

rdd.map(x=>x).saveAsTextFile("hdfs:ip//data/storage/20181010/"+(new Date()))  //new Date()自行转化

}



ssc.start()

ssc.awaitermination()


现在hsfs中目录显示为

   /data/storage/20181010/201810100708223

   /data/storage/20181010/201810100708460


达到自己想要的结果,根据streaming 间隔时间生成文件夹,并其中包含文件。

相关文章
|
2月前
|
消息中间件 分布式计算 NoSQL
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
44 0
|
2月前
|
消息中间件 存储 分布式计算
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
98 0
|
28天前
|
分布式计算 流计算 Spark
【赵渝强老师】Spark Streaming中的DStream
本文介绍了Spark Streaming的核心概念DStream,即离散流。DStream通过时间间隔将连续的数据流转换为一系列不连续的RDD,再通过Transformation进行转换,实现流式数据的处理。文中以MyNetworkWordCount程序为例,展示了DStream生成RDD的过程,并附有视频讲解。
|
2月前
|
消息中间件 分布式计算 Kafka
大数据-102 Spark Streaming Kafka ReceiveApproach DirectApproach 附带Producer、DStream代码案例
大数据-102 Spark Streaming Kafka ReceiveApproach DirectApproach 附带Producer、DStream代码案例
58 0
|
2月前
|
SQL 分布式计算 大数据
大数据-101 Spark Streaming DStream转换 窗口操作状态 跟踪操作 附带多个案例(一)
大数据-101 Spark Streaming DStream转换 窗口操作状态 跟踪操作 附带多个案例(一)
41 0
|
2月前
|
存储 分布式计算 大数据
大数据-101 Spark Streaming DStream转换 窗口操作状态 跟踪操作 附带多个案例(二)
大数据-101 Spark Streaming DStream转换 窗口操作状态 跟踪操作 附带多个案例(二)
49 0
|
2月前
|
SQL 分布式计算 大数据
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(一)
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(一)
32 0
|
2月前
|
SQL 分布式计算 大数据
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(二)
大数据-100 Spark 集群 Spark Streaming DStream转换 黑名单过滤的三种实现方式(二)
29 0
|
2月前
|
消息中间件 分布式计算 Kafka
大数据-99 Spark 集群 Spark Streaming DStream 文件数据流、Socket、RDD队列流
大数据-99 Spark 集群 Spark Streaming DStream 文件数据流、Socket、RDD队列流
31 0
|
2月前
|
消息中间件 分布式计算 Kafka
大数据-98 Spark 集群 Spark Streaming 基础概述 架构概念 执行流程 优缺点
大数据-98 Spark 集群 Spark Streaming 基础概述 架构概念 执行流程 优缺点
44 0