blocks|key|218904|text|fileStream生成NewHadoopRDDs的UnionRDD。sc.newAPIHadoopFile创建的NewHadoopRDDs的好处是它们的names被设置为它们的路径。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|218905|下面是你可以用这些知识做些什么的例子：|218906|def+namedTextFileStream(ssc:+StreamingContext,+directory:+String):+DStream[String]+=
++ssc.fileStream[LongWritable,+Text,+TextInputFormat](directory)
++++.transform(+rdd+=>
++++++new+UnionRDD(rdd.context,
++++++++rdd.dependencies.map(+dep+=>
++++++++++dep.rdd.asInstanceOf[RDD[(LongWritable,+Text)]].map(_._2.toString).setName(dep.rdd.name)
++++++++)
++++++)
++++)

def+transformByFile[U:+ClassTag](unionrdd:+RDD[String],
+++++++++++++++++++++++++++++++++transformFunc:+String+=>+RDD[String]+=>+RDD[U]):+RDD[U]+=+{
++new+UnionRDD(unionrdd.context,
++++unionrdd.dependencies.map{+dep+=>
++++++if+(dep.rdd.isEmpty)+None
++++++else+{
++++++++val+filename+=+dep.rdd.name
++++++++Some(
++++++++++transformFunc(filename)(dep.rdd.asInstanceOf[RDD[String]])
++++++++++++.setName(filename)
++++++++)
++++++}
++++}.flatten
++)
}

def+main(args:+Array[String])+=+{
++val+conf+=+new+SparkConf()
++++.setAppName("Process+by+file")
++++.setMaster("local[2]")

++val+ssc+=+new+StreamingContext(conf,+Seconds(30))

++val+dstream+=+namesTextFileStream(ssc,+"/some/directory")

++def+byFileTransformer(filename:+String)(rdd:+RDD[String]):+RDD[(String,+String)]+=
++++rdd.map(line+=>+(filename,+line))

++val+transformed+=+dstream.
++++transform(rdd+=>+transformByFile(rdd,+byFileTransformer))

++//+Do+some+stuff+with+transformed

++ssc.start()
++ssc.awaitTermination()
}|code-block|syntax|javascript|218907|entityMap^0|0|A|C|C|Q|8|Z|J|1L|C|25|4|0|0|0^^$0|@$1|2|3|4|5|6|7|O|8|@$9|P|A|Q|B|C]|$9|R|A|S|B|C]|$9|T|A|U|B|C]|$9|V|A|W|B|C]|$9|X|A|Y|B|C]|$9|Z|A|10|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|11|8|@]|D|@]|E|$]]|$1|H|3|I|5|J|7|12|8|@]|D|@]|E|$K|L]]|$1|M|3|-4|5|6|7|13|8|@]|D|@]|E|$]]]|N|$]]

<code>fileStream</code> produces <code>UnionRDD</code> of <code>NewHadoopRDD</code>s. The good part about <code>NewHadoopRDD</code>s created by <code>sc.newAPIHadoopFile</code> is that their <code>name</code>s are set to their paths.

Here's the example of what you can do with that knowledge:

<pre><code>def namedTextFileStream(ssc: StreamingContext, directory: String): DStream[String] =
 ssc.fileStream[LongWritable, Text, TextInputFormat](directory)
 .transform( rdd =&gt;
 new UnionRDD(rdd.context,
 rdd.dependencies.map( dep =&gt;
 dep.rdd.asInstanceOf[RDD[(LongWritable, Text)]].map(_._2.toString).setName(dep.rdd.name)
 )
 )
 )

def transformByFile[U: ClassTag](unionrdd: RDD[String],
 transformFunc: String =&gt; RDD[String] =&gt; RDD[U]): RDD[U] = {
 new UnionRDD(unionrdd.context,
 unionrdd.dependencies.map{ dep =&gt;
 if (dep.rdd.isEmpty) None
 else {
 val filename = dep.rdd.name
 Some(
 transformFunc(filename)(dep.rdd.asInstanceOf[RDD[String]])
 .setName(filename)
 )
 }
 }.flatten
 )
}

def main(args: Array[String]) = {
 val conf = new SparkConf()
 .setAppName("Process by file")
 .setMaster("local[2]")

 val ssc = new StreamingContext(conf, Seconds(30))

 val dstream = namesTextFileStream(ssc, "/some/directory")

 def byFileTransformer(filename: String)(rdd: RDD[String]): RDD[(String, String)] =
 rdd.map(line =&gt; (filename, line))

 val transformed = dstream.
 transform(rdd =&gt; transformByFile(rdd, byFileTransformer))

 // Do some stuff with transformed

 ssc.start()
 ssc.awaitTermination()
}
</code></pre>

blocks|key|440994|text|对于那些想要一些Java代码而不是Scala的人：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|440995|JavaPairInputDStream<LongWritable,+Text>+textFileStream+=+
++++++++jsc.fileStream(
++++++++++++inputPath,+
++++++++++++LongWritable.class,+
++++++++++++Text.class,
++++++++++++TextInputFormat.class,+
++++++++++++FileInputDStream::defaultFilter,
++++++++++++false
++++++++);
JavaDStream<Tuple2<String,+String>>+namedTextFileStream+=+textFileStream.transform((pairRdd,+time)+->+{
++++++++UnionRDD<Tuple2<LongWritable,+Text>>+rdd+=+(UnionRDD<Tuple2<LongWritable,+Text>>)+pairRdd.rdd();
++++++++List<RDD<Tuple2<LongWritable,+Text>>>+deps+=+JavaConverters.seqAsJavaListConverter(rdd.rdds()).asJava();
++++++++List<RDD<Tuple2<String,+String>>>+collectedRdds+=+deps.stream().map(+depRdd+->+{
++++++++++++if+(depRdd.isEmpty())+{
++++++++++++++++return+null;
++++++++++++}
++++++++++++JavaRDD<Tuple2<LongWritable,+Text>>+depJavaRdd+=+depRdd.toJavaRDD();
++++++++++++String+filename+=+depRdd.name();
++++++++++++JavaPairRDD<String,+String>+newDep+=+JavaPairRDD.fromJavaRDD(depJavaRdd).mapToPair(t+->+new+Tuple2<String,+String>(filename,+t._2().toString())).setName(filename);
++++++++++++return+newDep.rdd();
++++++++}).filter(t+->+t+!=+null).collect(Collectors.toList());
++++++++Seq<RDD<Tuple2<String,+String>>>+rddSeq+=+JavaConverters.asScalaBufferConverter(collectedRdds).asScala().toIndexedSeq();
++++++++ClassTag<Tuple2<String,+String>>+classTag+=+scala.reflect.ClassTag$.MODULE$.apply(Tuple2.class);
++++++++return+new+UnionRDD<Tuple2<String,+String>>(rdd.sparkContext(),+rddSeq,+classTag).toJavaRDD();
});|code-block|syntax|javascript|440996|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|I|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|J|8|@]|9|@]|A|$E|F]]|$1|G|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|H|$]]

For those that want some Java code instead of Scala:

<pre><code>JavaPairInputDStream&lt;LongWritable, Text&gt; textFileStream = 
 jsc.fileStream(
 inputPath, 
 LongWritable.class, 
 Text.class,
 TextInputFormat.class, 
 FileInputDStream::defaultFilter,
 false
 );
JavaDStream&lt;Tuple2&lt;String, String&gt;&gt; namedTextFileStream = textFileStream.transform((pairRdd, time) -&gt; {
 UnionRDD&lt;Tuple2&lt;LongWritable, Text&gt;&gt; rdd = (UnionRDD&lt;Tuple2&lt;LongWritable, Text&gt;&gt;) pairRdd.rdd();
 List&lt;RDD&lt;Tuple2&lt;LongWritable, Text&gt;&gt;&gt; deps = JavaConverters.seqAsJavaListConverter(rdd.rdds()).asJava();
 List&lt;RDD&lt;Tuple2&lt;String, String&gt;&gt;&gt; collectedRdds = deps.stream().map( depRdd -&gt; {
 if (depRdd.isEmpty()) {
 return null;
 }
 JavaRDD&lt;Tuple2&lt;LongWritable, Text&gt;&gt; depJavaRdd = depRdd.toJavaRDD();
 String filename = depRdd.name();
 JavaPairRDD&lt;String, String&gt; newDep = JavaPairRDD.fromJavaRDD(depJavaRdd).mapToPair(t -&gt; new Tuple2&lt;String, String&gt;(filename, t._2().toString())).setName(filename);
 return newDep.rdd();
 }).filter(t -&gt; t != null).collect(Collectors.toList());
 Seq&lt;RDD&lt;Tuple2&lt;String, String&gt;&gt;&gt; rddSeq = JavaConverters.asScalaBufferConverter(collectedRdds).asScala().toIndexedSeq();
 ClassTag&lt;Tuple2&lt;String, String&gt;&gt; classTag = scala.reflect.ClassTag$.MODULE$.apply(Tuple2.class);
 return new UnionRDD&lt;Tuple2&lt;String, String&gt;&gt;(rdd.sparkContext(), rddSeq, classTag).toJavaRDD();
});
</code></pre>

blocks|key|225798|text|或者，通过修改FileInputDStream，而不是将文件内容加载到RDD中，它只需从文件名创建RDD即可。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|225799|如果您实际上不想将数据本身读取到RDD中，或者希望将文件名作为您的步骤之一传递给外部命令，则这会提高性能。|225800|只需更改filesToRDD(..)因此，它对文件名进行RDD，而不是将数据加载到RDD中。|225801|请参阅：https://github.com/HASTE-project/bin-packing-paper/blob/master/spark/spark-scala-cellprofiler/src/main/scala/FileInputDStream2.scala#L278|offset|length|225802|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/HASTE-project/bin-packing-paper/blob/master/spark/spark-scala-cellprofiler/src/main/scala/FileInputDStream2.scala#L278^0|0|0|0|4|3T|0|0^^$0|@$1|2|3|4|5|6|7|R|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|S|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|T|8|@]|9|@]|A|$]]|$1|F|3|G|5|6|7|U|8|@]|9|@$H|V|I|W|1|X]]|A|$]]|$1|J|3|-4|5|6|7|Y|8|@]|9|@]|A|$]]]|K|$L|$5|M|N|O|A|$P|Q]]]]

Alternatively, by modifying FileInputDStream so that rather than loading the contents of the files into the RDD, it simply creates an RDD from the filenames.

This gives a performance boost if you don't actually want to read the data itself into the RDD, or want to pass filenames to an external command as one of your steps.

Simply change filesToRDD(..) so that it makes an RDD of the filenames, rather than loading the data into the RDD.

See: <a href="https://github.com/HASTE-project/bin-packing-paper/blob/master/spark/spark-scala-cellprofiler/src/main/scala/FileInputDStream2.scala#L278" rel="nofollow noreferrer">https://github.com/HASTE-project/bin-packing-paper/blob/master/spark/spark-scala-cellprofiler/src/main/scala/FileInputDStream2.scala#L278</a>

Spark streaming <code>textFileStream</code> and <code>fileStream</code> can monitor a directory and process the new files in a Dstream RDD.

How to get the file names that are being processed by the DStream RDD at that particular interval?

Spark streaming DStream RDD to get file name

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

Spark streaming textFileStream和fileStream可以监控目录并处理Dstream RDD中的新文件。如何获取DStream RDD在该特定时间间隔正在处理的文件名？

问Spark streaming DStream RDD获取文件名
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Spark streaming DStream RDD获取文件名EN