Spark Streaming 不显示任何关于应用程序的记录 UI
Spark Streaming does not display any record on application UI
我是 spark 的新手，我正在尝试 运行 一个简单的 spark 流应用程序，它从 csv 文件读取数据并显示它。似乎 Spark Streaming 有效，但它仍然在 Streaming UI 上显示“0”条记录 application.Here 是我的代码：
public class App {
public static void main(String[] args) throws Exception {
// Get an instance of spark-conf, required to build the spark session
SparkConf conf = new SparkConf().setAppName("StreamingExample").setMaster("local");
JavaStreamingContext jsc = new JavaStreamingContext(conf, new Duration(3000));
//JavaSparkContext ssc= new JavaSparkContext(conf);
jsc.checkpoint("checkpoint");
System.out.println("Session created");
JavaDStream < String > lines = jsc.textFileStream("C:\Users\Areeha\eclipse-workspace\learnspark\src\main\java\com\example\learnspark");
lines.print();
lines.foreachRDD(rdd - > rdd.foreach(x - > System.out.println(x)));
JavaPairDStream < LongWritable, Text > streamedFile = jsc.fileStream("C:\Users\Areeha\eclipse-workspace\learnspark\src\main\java\com\example\learnspark", LongWritable.class, Text.class, TextInputFormat.class);
streamedFile.print();
System.out.println("File loaded!");
System.out.println(streamedFile.count());
System.out.println(lines.count());
jsc.start();
try {
jsc.awaitTermination();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
这是我在控制台上得到的：
Using Spark 's default log4j profile: org/apache/spark/log4j-defaults.properties
19 / 11 / 21 09: 24: 50 INFO SparkContext: Running Spark version 2.4 .4
19 / 11 / 21 09: 24: 50 WARN NativeCodeLoader: Unable to load native - hadoop library
for your platform...using builtin - java classes where applicable
19 / 11 / 21 09: 24: 50 INFO SparkContext: Submitted application: StreamingExample
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing view acls to: Areeha
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing modify acls to: Areeha
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing view acls groups to:
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing modify acls groups to:
19 / 11 / 21 09: 24: 50 INFO SecurityManager: SecurityManager: authentication disabled;
ui acls disabled;
users with view permissions: Set(Areeha);
groups with view permissions: Set();
users with modify permissions: Set(Areeha);
groups with modify permissions: Set()
19 / 11 / 21 09: 24: 51 INFO Utils: Successfully started service 'sparkDriver'
on port 57635.
19 / 11 / 21 09: 24: 51 INFO SparkEnv: Registering MapOutputTracker
19 / 11 / 21 09: 24: 51 INFO SparkEnv: Registering BlockManagerMaster
19 / 11 / 21 09: 24: 51 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper
for getting topology information
19 / 11 / 21 09: 24: 51 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
19 / 11 / 21 09: 24: 51 INFO DiskBlockManager: Created local directory at C: \Users\ Areeha\ AppData\ Local\ Temp\ blockmgr - 9 d8ba7c2 - 3 b21 - 419 c - 8711 - d85f7d1704a1
19 / 11 / 21 09: 24: 51 INFO MemoryStore: MemoryStore started with capacity 1443.6 MB
19 / 11 / 21 09: 24: 51 INFO SparkEnv: Registering OutputCommitCoordinator
19 / 11 / 21 09: 24: 52 INFO Utils: Successfully started service 'SparkUI'
on port 4040.
19 / 11 / 21 09: 24: 52 INFO SparkUI: Bound SparkUI to 0.0 .0 .0, and started at http: //192.168.2.8:4040
19 / 11 / 21 09: 24: 52 INFO Executor: Starting executor ID driver on host localhost
19 / 11 / 21 09: 24: 52 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService'
on port 57648.
19 / 11 / 21 09: 24: 52 INFO NettyBlockTransferService: Server created on 192.168 .2 .8: 57648
19 / 11 / 21 09: 24: 52 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy
for block replication policy
19 / 11 / 21 09: 24: 52 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 INFO BlockManagerMasterEndpoint: Registering block manager 192.168 .2 .8: 57648 with 1443.6 MB RAM, BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 WARN StreamingContext: spark.master should be set as local[n], n > 1 in local mode
if you have receivers to get data, otherwise Spark jobs will not get resources to process the received data.
Session created
19 / 11 / 21 09: 24: 52 INFO FileInputDStream: Duration
for remembering RDDs set to 60000 ms
for org.apache.spark.streaming.dstream.FileInputDStream @14151bc5
19 / 11 / 21 09: 24: 52 INFO FileInputDStream: Duration
for remembering RDDs set to 60000 ms
for org.apache.spark.streaming.dstream.FileInputDStream @151335cb
File loaded!
org.apache.spark.streaming.api.java.JavaDStream @46d8f407
org.apache.spark.streaming.api.java.JavaDStream @2788d0fe
19 / 11 / 21 09: 24: 53 INFO FileBasedWriteAheadLog_ReceivedBlockTracker: Recovered 4 write ahead log files from file: /C:/Users / Areeha / eclipse - workspace / learnspark / checkpoint / receivedBlockMetadata
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Remember interval = 60000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Initialized and validated org.apache.spark.streaming.dstream.FileInputDStream @14151bc5
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Initialized and validated org.apache.spark.streaming.dstream.MappedDStream @528f8f8b
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream @4cbf4f53
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Remember interval = 60000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Initialized and validated org.apache.spark.streaming.dstream.FileInputDStream @14151bc5
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Initialized and validated org.apache.spark.streaming.dstream.MappedDStream @528f8f8b
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream @58d63b16
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Remember interval = 60000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Initialized and validated org.apache.spark.streaming.dstream.FileInputDStream @151335cb
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream @748e9b20
19 / 11 / 21 09: 24: 53 INFO RecurringTimer: Started timer
for JobGenerator at time 1574349894000
19 / 11 / 21 09: 24: 53 INFO JobGenerator: Started JobGenerator at 1574349894000 ms
19 / 11 / 21 09: 24: 53 INFO JobScheduler: Started JobScheduler
19 / 11 / 21 09: 24: 53 INFO StreamingContext: StreamingContext started
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: Finding new files took 9 ms
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: New files at time 1574349894000 ms:
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: Finding new files took 3 ms
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: New files at time 1574349894000 ms:
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Added jobs
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobGenerator: Checkpointing graph
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO DStreamGraph: Updating checkpoint data
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Starting job streaming job 1574349894000 ms .0 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO DStreamGraph: Updated checkpoint data
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO CheckpointWriter: Submitted checkpoint of time 1574349894000 ms to writer queue
19 / 11 / 21 09: 24: 54 INFO CheckpointWriter: Saving checkpoint
for time 1574349894000 ms to file 'file:/C:/Users/Areeha/eclipse-workspace/learnspark/checkpoint/checkpoint-1574349894000'
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
Time: 1574349894000 ms
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Finished job streaming job 1574349894000 ms .0 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Starting job streaming job 1574349894000 ms .1 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO SparkContext: Starting job: foreach at App.java: 79
19 / 11 / 21 09: 24: 54 INFO DAGScheduler: Job 0 finished: foreach at App.java: 79, took 0.002286 s
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Finished job streaming job 1574349894000 ms .1 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Starting job streaming job 1574349894000 ms .2 from job set of time 1574349894000 ms
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
Time: 1574349894000 ms
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
下面出现在我的 Streaming UI 应用程序中：enter image description here
我不知道我做错了什么。它既不显示任何内容也不向其添加任何记录。我之前已经指定了确切的 csv 文件，但它不起作用，所以我尝试提供包含 csv.Does 的整个文件夹的路径，有人知道我错过了什么吗？
提前致谢。
TextFileStream 不使用 Receiver 线程，因此不会像其他来源那样在 UI 中记录记录：
File Streams
For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as via StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass].
File streams do not require running a receiver so there is no need to allocate any cores for receiving file data.
来源：https://spark.apache.org/docs/2.3.1/streaming-custom-receivers.html
有人在 this JIRA ticket 上打开了一个 PR，更改了 Spark 逻辑，所以此信息但票证没有设置修复版本。
我通常做的是知道每批输入了多少条记录，是在 forEachRDD 中处理 RDD 时记录计数：
lines.forEachRDD( rdd -> {
// You might want to cache the rdd before counting if you are dealing with large RDDs
logger.debug(s"${rdd.count() records found")
})
编辑：关于你的文件没有被处理，你可能想在你的日志配置中设置 DEBUG 这个包 org.apache.spark.streaming.dstream.FileInputDStream 因为它说它“看到”了哪些文件以及为什么接受还是不接受（主要是时间戳太旧）
我是 spark 的新手，我正在尝试 运行 一个简单的 spark 流应用程序，它从 csv 文件读取数据并显示它。似乎 Spark Streaming 有效，但它仍然在 Streaming UI 上显示“0”条记录 application.Here 是我的代码：
public class App {
public static void main(String[] args) throws Exception {
// Get an instance of spark-conf, required to build the spark session
SparkConf conf = new SparkConf().setAppName("StreamingExample").setMaster("local");
JavaStreamingContext jsc = new JavaStreamingContext(conf, new Duration(3000));
//JavaSparkContext ssc= new JavaSparkContext(conf);
jsc.checkpoint("checkpoint");
System.out.println("Session created");
JavaDStream < String > lines = jsc.textFileStream("C:\Users\Areeha\eclipse-workspace\learnspark\src\main\java\com\example\learnspark");
lines.print();
lines.foreachRDD(rdd - > rdd.foreach(x - > System.out.println(x)));
JavaPairDStream < LongWritable, Text > streamedFile = jsc.fileStream("C:\Users\Areeha\eclipse-workspace\learnspark\src\main\java\com\example\learnspark", LongWritable.class, Text.class, TextInputFormat.class);
streamedFile.print();
System.out.println("File loaded!");
System.out.println(streamedFile.count());
System.out.println(lines.count());
jsc.start();
try {
jsc.awaitTermination();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
这是我在控制台上得到的：
Using Spark 's default log4j profile: org/apache/spark/log4j-defaults.properties
19 / 11 / 21 09: 24: 50 INFO SparkContext: Running Spark version 2.4 .4
19 / 11 / 21 09: 24: 50 WARN NativeCodeLoader: Unable to load native - hadoop library
for your platform...using builtin - java classes where applicable
19 / 11 / 21 09: 24: 50 INFO SparkContext: Submitted application: StreamingExample
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing view acls to: Areeha
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing modify acls to: Areeha
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing view acls groups to:
19 / 11 / 21 09: 24: 50 INFO SecurityManager: Changing modify acls groups to:
19 / 11 / 21 09: 24: 50 INFO SecurityManager: SecurityManager: authentication disabled;
ui acls disabled;
users with view permissions: Set(Areeha);
groups with view permissions: Set();
users with modify permissions: Set(Areeha);
groups with modify permissions: Set()
19 / 11 / 21 09: 24: 51 INFO Utils: Successfully started service 'sparkDriver'
on port 57635.
19 / 11 / 21 09: 24: 51 INFO SparkEnv: Registering MapOutputTracker
19 / 11 / 21 09: 24: 51 INFO SparkEnv: Registering BlockManagerMaster
19 / 11 / 21 09: 24: 51 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper
for getting topology information
19 / 11 / 21 09: 24: 51 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
19 / 11 / 21 09: 24: 51 INFO DiskBlockManager: Created local directory at C: \Users\ Areeha\ AppData\ Local\ Temp\ blockmgr - 9 d8ba7c2 - 3 b21 - 419 c - 8711 - d85f7d1704a1
19 / 11 / 21 09: 24: 51 INFO MemoryStore: MemoryStore started with capacity 1443.6 MB
19 / 11 / 21 09: 24: 51 INFO SparkEnv: Registering OutputCommitCoordinator
19 / 11 / 21 09: 24: 52 INFO Utils: Successfully started service 'SparkUI'
on port 4040.
19 / 11 / 21 09: 24: 52 INFO SparkUI: Bound SparkUI to 0.0 .0 .0, and started at http: //192.168.2.8:4040
19 / 11 / 21 09: 24: 52 INFO Executor: Starting executor ID driver on host localhost
19 / 11 / 21 09: 24: 52 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService'
on port 57648.
19 / 11 / 21 09: 24: 52 INFO NettyBlockTransferService: Server created on 192.168 .2 .8: 57648
19 / 11 / 21 09: 24: 52 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy
for block replication policy
19 / 11 / 21 09: 24: 52 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 INFO BlockManagerMasterEndpoint: Registering block manager 192.168 .2 .8: 57648 with 1443.6 MB RAM, BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, 192.168 .2 .8, 57648, None)
19 / 11 / 21 09: 24: 52 WARN StreamingContext: spark.master should be set as local[n], n > 1 in local mode
if you have receivers to get data, otherwise Spark jobs will not get resources to process the received data.
Session created
19 / 11 / 21 09: 24: 52 INFO FileInputDStream: Duration
for remembering RDDs set to 60000 ms
for org.apache.spark.streaming.dstream.FileInputDStream @14151bc5
19 / 11 / 21 09: 24: 52 INFO FileInputDStream: Duration
for remembering RDDs set to 60000 ms
for org.apache.spark.streaming.dstream.FileInputDStream @151335cb
File loaded!
org.apache.spark.streaming.api.java.JavaDStream @46d8f407
org.apache.spark.streaming.api.java.JavaDStream @2788d0fe
19 / 11 / 21 09: 24: 53 INFO FileBasedWriteAheadLog_ReceivedBlockTracker: Recovered 4 write ahead log files from file: /C:/Users / Areeha / eclipse - workspace / learnspark / checkpoint / receivedBlockMetadata
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Remember interval = 60000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Initialized and validated org.apache.spark.streaming.dstream.FileInputDStream @14151bc5
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Initialized and validated org.apache.spark.streaming.dstream.MappedDStream @528f8f8b
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream @4cbf4f53
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Remember interval = 60000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Initialized and validated org.apache.spark.streaming.dstream.FileInputDStream @14151bc5
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO MappedDStream: Initialized and validated org.apache.spark.streaming.dstream.MappedDStream @528f8f8b
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream @58d63b16
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Remember interval = 60000 ms
19 / 11 / 21 09: 24: 53 INFO FileInputDStream: Initialized and validated org.apache.spark.streaming.dstream.FileInputDStream @151335cb
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Slide time = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Storage level = Serialized 1 x Replicated
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Checkpoint interval = null
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Remember interval = 3000 ms
19 / 11 / 21 09: 24: 53 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream @748e9b20
19 / 11 / 21 09: 24: 53 INFO RecurringTimer: Started timer
for JobGenerator at time 1574349894000
19 / 11 / 21 09: 24: 53 INFO JobGenerator: Started JobGenerator at 1574349894000 ms
19 / 11 / 21 09: 24: 53 INFO JobScheduler: Started JobScheduler
19 / 11 / 21 09: 24: 53 INFO StreamingContext: StreamingContext started
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: Finding new files took 9 ms
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: New files at time 1574349894000 ms:
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: Finding new files took 3 ms
19 / 11 / 21 09: 24: 54 INFO FileInputDStream: New files at time 1574349894000 ms:
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Added jobs
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobGenerator: Checkpointing graph
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO DStreamGraph: Updating checkpoint data
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Starting job streaming job 1574349894000 ms .0 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO DStreamGraph: Updated checkpoint data
for time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO CheckpointWriter: Submitted checkpoint of time 1574349894000 ms to writer queue
19 / 11 / 21 09: 24: 54 INFO CheckpointWriter: Saving checkpoint
for time 1574349894000 ms to file 'file:/C:/Users/Areeha/eclipse-workspace/learnspark/checkpoint/checkpoint-1574349894000'
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
Time: 1574349894000 ms
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Finished job streaming job 1574349894000 ms .0 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Starting job streaming job 1574349894000 ms .1 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO SparkContext: Starting job: foreach at App.java: 79
19 / 11 / 21 09: 24: 54 INFO DAGScheduler: Job 0 finished: foreach at App.java: 79, took 0.002286 s
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Finished job streaming job 1574349894000 ms .1 from job set of time 1574349894000 ms
19 / 11 / 21 09: 24: 54 INFO JobScheduler: Starting job streaming job 1574349894000 ms .2 from job set of time 1574349894000 ms
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
Time: 1574349894000 ms
-- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -- -
下面出现在我的 Streaming UI 应用程序中：enter image description here
我不知道我做错了什么。它既不显示任何内容也不向其添加任何记录。我之前已经指定了确切的 csv 文件，但它不起作用，所以我尝试提供包含 csv.Does 的整个文件夹的路径，有人知道我错过了什么吗？ 提前致谢。
TextFileStream 不使用 Receiver 线程，因此不会像其他来源那样在 UI 中记录记录：
File Streams
For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as via StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass].
File streams do not require running a receiver so there is no need to allocate any cores for receiving file data.
来源：https://spark.apache.org/docs/2.3.1/streaming-custom-receivers.html
有人在 this JIRA ticket 上打开了一个 PR，更改了 Spark 逻辑，所以此信息但票证没有设置修复版本。
我通常做的是知道每批输入了多少条记录，是在 forEachRDD 中处理 RDD 时记录计数：
lines.forEachRDD( rdd -> {
// You might want to cache the rdd before counting if you are dealing with large RDDs
logger.debug(s"${rdd.count() records found")
})
编辑：关于你的文件没有被处理，你可能想在你的日志配置中设置 DEBUG 这个包 org.apache.spark.streaming.dstream.FileInputDStream 因为它说它“看到”了哪些文件以及为什么接受还是不接受（主要是时间戳太旧）