【问题标题】：Unable to write data from TCP port to HDFS using Spark Streaming无法使用 Spark Streaming 将数据从 TCP 端口写入 HDFS
【发布时间】：2014-06-25 13:00:52
【问题描述】：

我正在尝试从 TCP 端口流式传输数据并使用 Spark-Streaming 将数据加载到 HDFS。

文件正在 HDFS 中创建，但它们都是空的。但 Spark Streaming 控制台显示从 TCP 端口读取数据。

我在 Spark 0.9.0、0.9.1 和 1.0 中使用 CDH-5 中的 Scala-Shell 进行了尝试。我在另一个终端中执行了“nc -lk 9993”来流式传输数据。

以下是代码，请告知如何解决此问题。谢谢。

import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.api.java.function._
import org.apache.spark.streaming._
import org.apache.spark.streaming.api._
import org.apache.spark.streaming.StreamingContext._
import StreamingContext._

val ssc8 = new StreamingContext("local", "NetworkWordCount", Seconds(1))
val lines8 = ssc8.socketTextStream("localhost", 9993)

val words8 = lines8.flatMap(_.split(" "))


val pairs8 = words8.map(word => (word, 1))
val wordCounts8 = pairs8.reduceByKey(_ + _)

wordCounts8.saveAsTextFiles("hdfs://Node1:8020/user/root/Spark8")

wordCounts8.print()

ssc8.start()

已添加---------------------------------------

我在下面提供了日志和 HDFS 文件 -

HDFS  Output Files
--------------------

-rw-r--r--   3 user1 user1          0 2014-06-26 09:19 /user/user1/SparkV/_SUCCESS
-rw-r--r--   3 user1 user1         0 2014-06-26 09:19 /user/user1/SparkV/part-00000
-rw-r--r--   3 user1 user1          0 2014-06-26 09:19 /user/user1/SparkV/part-00001




Spark-Shell Console Log
---------------------


-------------------------------------------
Time: 1403789836000 ms
-------------------------------------------
(f,3)
(fsd,2)
(sdf,2)
(fds,1)
(sd,3)

14/06/26 09:37:16 INFO scheduler.JobScheduler: Finished job streaming job 1403789836000 ms.1 from job set of time 1403789836000 ms
14/06/26 09:37:16 INFO storage.MemoryStore: ensureFreeSpace(8) called with curMem=327, maxMem=286339891
14/06/26 09:37:16 INFO storage.MemoryStore: Block input-0-1403789836000 stored as bytes to memory (size 8.0 B, free 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerInfo: Added input-0-1403789836000 in memory on localhost:49784 (size: 8.0 B, free: 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerMaster: Updated info of block input-0-1403789836000
14/06/26 09:37:16 WARN storage.BlockManager: Block input-0-1403789836000 already exists on this machine; not re-adding it
14/06/26 09:37:16 INFO receiver.BlockGenerator: Pushed block input-0-1403789836000
14/06/26 09:37:16 INFO storage.MemoryStore: ensureFreeSpace(15) called with curMem=335, maxMem=286339891
14/06/26 09:37:16 INFO storage.MemoryStore: Block input-0-1403789836200 stored as bytes to memory (size 15.0 B, free 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerInfo: Added input-0-1403789836200 in memory on localhost:49784 (size: 15.0 B, free: 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerMaster: Updated info of block input-0-1403789836200
14/06/26 09:37:16 WARN storage.BlockManager: Block input-0-1403789836200 already exists on this machine; not re-adding it
14/06/26 09:37:16 INFO receiver.BlockGenerator: Pushed block input-0-1403789836200
14/06/26 09:37:16 INFO storage.MemoryStore: ensureFreeSpace(8) called with curMem=350, maxMem=286339891
14/06/26 09:37:16 INFO storage.MemoryStore: Block input-0-1403789836400 stored as bytes to memory (size 8.0 B, free 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerInfo: Added input-0-1403789836400 in memory on localhost:49784 (size: 8.0 B, free: 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerMaster: Updated info of block input-0-1403789836400
14/06/26 09:37:16 WARN storage.BlockManager: Block input-0-1403789836400 already exists on this machine; not re-adding it
14/06/26 09:37:16 INFO receiver.BlockGenerator: Pushed block input-0-1403789836400
14/06/26 09:37:16 INFO storage.MemoryStore: ensureFreeSpace(9) called with curMem=358, maxMem=286339891
14/06/26 09:37:16 INFO storage.MemoryStore: Block input-0-1403789836600 stored as bytes to memory (size 9.0 B, free 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerInfo: Added input-0-1403789836600 in memory on localhost:49784 (size: 9.0 B, free: 273.1 MB)
14/06/26 09:37:16 INFO storage.BlockManagerMaster: Updated info of block input-0-1403789836600
14/06/26 09:37:16 WARN storage.BlockManager: Block input-0-1403789836600 already exists on this machine; not re-adding it
14/06/26 09:37:16 INFO receiver.BlockGenerator: Pushed block input-0-1403789836600
14/06/26 09:37:17 INFO storage.MemoryStore: ensureFreeSpace(14) called with curMem=367, maxMem=286339891
14/06/26 09:37:17 INFO storage.MemoryStore: Block input-0-1403789836800 stored as bytes to memory (size 14.0 B, free 273.1 MB)
14/06/26 09:37:17 INFO storage.BlockManagerInfo: Added input-0-1403789836800 in memory on localhost:49784 (size: 14.0 B, free: 273.1 MB)
14/06/26 09:37:17 INFO storage.BlockManagerMaster: Updated info of block input-0-1403789836800
14/06/26 09:37:17 WARN storage.BlockManager: Block input-0-1403789836800 already exists on this machine; not re-adding it
14/06/26 09:37:17 INFO receiver.BlockGenerator: Pushed block input-0-1403789836800
14/06/26 09:37:18 INFO scheduler.ReceiverTracker: Stream 0 received 6 blocks
14/06/26 09:37:18 INFO scheduler.JobScheduler: Added jobs for time 1403789838000 ms

【问题讨论】：

标签： scala streaming hdfs apache-spark

【解决方案1】：

乍一看，我的猜测是您应该尝试本地[4] 而不是本地，这样 Spark 可以安排更多任务。

【讨论】：

谢谢。我尝试同时使用 local[2] 和 local[4]，但它仍然没有写入 HDFS。但是 - 1. 以前控制台上没有显示字数，现在可以了。 2.现在连同目录中的 _SUCESS 文件，part-* 文件也被创建 - 但都是空的。另外 - 我没有使用 ssc8.awaitTermination() 方法。而且，当我使用 ^D 或终止 netcat 功能时，仍然没有用。我在这里错过了什么吗？另外，我在原始帖子中附加了额外的日志和 HDFS 输出文件。
很高兴我们取得了一些进展，您有什么理由不使用 awaitTermination() 吗？在这种情况下，我的猜测是您的驱动程序可能会提前退出。
嗨，一旦指定了 start() 方法，我将无法使用 awaitTermination()，流式日志从控制台开始，我无法调用方法 awaitTermination()。如果我打开一个新终端，打开 spark shell 并调用 var.awaitTermination() 函数，它什么也不做。如果我的理解不正确或者我在这里做错了什么，请提供帮助。
如果您通过使用提交客户端提交应用程序驱动程序完全在集群内部运行应用程序。然后以下命令应该可以工作 ./bin/spark-class org.apache.spark.deploy.Client kill

【解决方案2】：

试试

wordCounts8.saveAsTextFiles("hdfs://Node1:8020/user/root/Spark8", "log")

=========== 或

在“Spark8”之后添加一些时间戳

wordCounts8.saveAsTextFiles("hdfs://Node1:8020/user/root/Spark8" + System.currentTimeMillis().toString())

=========== spark 1.3 对我有用，看看是否对你有用

【讨论】：

【解决方案3】：

我遇到了同样的问题。

尝试运行

hadoop fs -cat hdfs://Node1:8020/user/root/Spark8

（hadoop 命令对你来说可能不同。对我来说，我必须使用 /a/bin/hadoop 访问它，但这取决于你的设置）

看看这是否返回：

cat: `hdfs://Node1:8020/user/root/Spark8': Is a directory

如果是这样，那么正如您在 cmets 中所说，您应该能够在该目录中看到一个 _SUCCESS 文件以及一些 part-* 文件。

至此，我的问题解决了。但是您似乎在写入 HDFS 时遇到了其他问题。

至于为什么您的文件仍然是空的，我建议切换到 Spark1.4.0，因为使用 CDH5.4 可能会更好。此外，如果您在 HDFS 权限方面遇到问题，则必须执行

hadoop dfs -chmod -R 0777 /your_hdfs_folder

为了有写权限。

【讨论】：