阿帕奇星火室壁运动样品不工作 [英] Apache Spark Kinesis Sample not working

问题描述

我试图运行<一个href=\"https://github.com/apache/spark/blob/master/extras/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.java\"相对=nofollow> JavaKinesisWordCountASL 例子。

这个例子似乎连接到我的Kinesis流和从流中获取数据（如下面的日志中）。然而，火花不调用传递给本例中的unionStreams.flatMap方法的呼叫功能，并且不打印任何单词计数

The example seem to connect to my Kinesis Stream and gets data from the stream (as shown in the log below). However, Sparks does not invoke the call function passed to the unionStreams.flatMap method in the example and does not prints any wordcount.

我曾尝试运行使用包括Java 8和Java 7，我运行它的Ubuntu实例。同样的例子适用于我的MacBook。

I have tried running using both Java 8 and Java 7. I am running it on an ubuntu instance. The same example works on my macbook.

14/11/15 1时59分42秒INFO scheduler.ReceiverTracker：流1接收0块
14/11/15 1时59分42秒INFO storage.MemoryStore：ensureFreeSpace（264）调用curMem = 3512，MAXMEM = 938244833
14/11/15 1时59分42秒INFO storage.MemoryStore：阻止输入0-1416016781800存储在内存中的值（估计大小264.0 B，免费894.8 MB）
14/11/15 1时59分42秒INFO storage.BlockManagerInfo：在内存中增加了输入0-1416016781800对IP-10-80-91-13.ec2.internal：39149（尺寸：264.0 B，免费：894.8 MB）
14/11/15 1时59分42秒INFO storage.BlockManagerMaster：块输入0-1416016781800的更新信息
14/11/15 1时59分42秒INFO scheduler.JobScheduler：时间1416016782000女士补充工作
14/11/15 1时59分42秒INFO network.SendingConnection：启动连接[IP-10-80-91-13.ec2.internal / 10.80.91.13：39149]
14/11/15 1时59分42秒INFO network.SendingConnection：连接到[IP-10-80-91-13.ec2.internal / 10.80.91.13：39149]，1未决的消息
14/11/15 1时59分42秒INFO network.ConnectionManager：从接受的连接[IP-10-80-91-13.ec2.internal / 10.80.91.13：56700]
14/11/15 1时59分42秒WARN storage.BlockManager：阻止输入0-1416016781800已经存在这台机器上;不重新添加
14/11/15 1时59分42秒INFO receiver.BlockGenerator：推块输入0-1416016781800
14/11/15一时59分43秒INFO storage.MemoryStore：ensureFreeSpace（256）调用curMem = 3776，MAXMEM = 938244833
14/11/15一时59分43秒INFO storage.MemoryStore：阻止输入0-1416016782800存储在内存中的值（估计大小256.0 B，免费894.8 MB）
14/11/15一时59分43秒INFO storage.BlockManagerInfo：在内存中增加了输入0-1416016782800对IP-10-80-91-13.ec2.internal：39149（尺寸：256.0 B，免费：894.8 MB）
14/11/15一时59分43秒INFO storage.BlockManagerMaster：块输入0-1416016782800的更新信息
14/11/15一时59分43秒WARN storage.BlockManager：阻止输入0-1416016782800已经存在这台机器上;不重新添加
14/11/15一时59分43秒INFO receiver.BlockGenerator：推块输入0-1416016782800
14/11/15 1时59分44秒INFO scheduler.ReceiverTracker：0流收到2个街区
14/11/15 1时59分44秒INFO scheduler.ReceiverTracker：流1接收0块
14/11/15 1时59分44秒INFO scheduler.JobScheduler：时间1416016784000女士补充工作
14/11/15 1时59分46秒INFO scheduler.ReceiverTracker：0流收到0块
14/11/15 1时59分46秒INFO scheduler.ReceiverTracker：流1接收0块
14/11/15 1时59分46秒INFO scheduler.JobScheduler：时间1416016786000女士补充工作
14/11/15 1时59分46秒INFO impl.CWPublisherRunnable：成功发布17基准。
14/11/15 1时59分46秒INFO storage.MemoryStore：ensureFreeSpace（248）调用curMem = 4032，MAXMEM = 938244833
14/11/15 1时59分46秒INFO storage.MemoryStore：阻止输入1-1416016786000存储在内存中的值（估计大小248.0 B，免费894.8 MB）
14/11/15 1时59分46秒INFO storage.BlockManagerInfo：在内存中增加了输入1-1416016786000对IP-10-80-91-13.ec2.internal：39149（尺寸：248.0 B，免费：894.8 MB）
14/11/15 1时59分46秒INFO storage.BlockManagerMaster：块输入1-1416016786000的更新信息
14/11/15 1时59分46秒WARN storage.BlockManager：阻止输入1-1416016786000已经存在这台机器上;不重新添加
14/11/15 1时59分46秒INFO receiver.BlockGenerator：推块输入1-1416016786000
14/11/15 1时59分46秒INFO impl.CWPublisherRunnable：成功发布14基准。
14/11/15 1点59分48秒INFO scheduler.ReceiverTracker：0流收到0块
14/11/15 1点59分48秒INFO storage.MemoryStore：ensureFreeSpace（264）调用curMem = 4280，MAXMEM = 938244833
14/11/15 1点59分48秒INFO scheduler.ReceiverTracker：流1接收1块
14/11/15 1点59分48秒INFO storage.MemoryStore：阻止输入0-1416016787800存储在内存中的值（估计大小264.0 B，免费894.8 MB）
14/11/15 1点59分48秒INFO storage.BlockManagerInfo：在内存中增加了输入0-1416016787800对IP-10-80-91-13.ec2.internal：39149（尺寸：264.0 B，免费：894.8 MB）
14/11/15 1点59分48秒INFO storage.BlockManagerMaster：块输入0-1416016787800的更新信息
14/11/15 1点59分48秒INFO scheduler.JobScheduler：时间1416016788000女士补充工作
14/11/15 1点59分48秒WARN storage.BlockManager：阻止输入0-1416016787800已经存在这台机器上;不重新添加
14/11/15 1点59分48秒INFO receiver.BlockGenerator：推块输入0-1416016787800
14/11/15 1时59分50秒INFO scheduler.ReceiverTracker：流0获得1块
14/11/15 1时59分50秒INFO scheduler.ReceiverTracker：流1接收0块
14/11/15 1时59分50秒INFO scheduler.JobScheduler：时间1416016790000女士补充工作
14/11/15 1点59分51秒INFO storage.MemoryStore：ensureFreeSpace（264）调用curMem = 4544，MAXMEM = 938244833
14/11/15 1点59分51秒INFO storage.MemoryStore：阻止输入0-1416016790800存储在内存中的值（估计大小264.0 B，免费894.8 MB）
14/11/15 1点59分51秒INFO storage.BlockManagerInfo：在内存中增加了输入0-1416016790800对IP-10-80-91-13.ec2.internal：39149（尺寸：264.0 B，免费：894.8 MB）
14/11/15 1点59分51秒INFO storage.BlockManagerMaster：块输入0-1416016790800的更新信息
14/11/15 1点59分51秒WARN storage.BlockManager：阻止输入0-1416016790800已经存在这台机器上;不重新添加
14/11/15 1点59分51秒INFO receiver.BlockGenerator：推块输入0-1416016790800

14/11/15 01:59:42 INFO scheduler.ReceiverTracker: Stream 1 received 0 blocks 14/11/15 01:59:42 INFO storage.MemoryStore: ensureFreeSpace(264) called with curMem=3512, maxMem=938244833 14/11/15 01:59:42 INFO storage.MemoryStore: Block input-0-1416016781800 stored as values in memory (estimated size 264.0 B, free 894.8 MB) 14/11/15 01:59:42 INFO storage.BlockManagerInfo: Added input-0-1416016781800 in memory on ip-10-80-91-13.ec2.internal:39149 (size: 264.0 B, free: 894.8 MB) 14/11/15 01:59:42 INFO storage.BlockManagerMaster: Updated info of block input-0-1416016781800 14/11/15 01:59:42 INFO scheduler.JobScheduler: Added jobs for time 1416016782000 ms 14/11/15 01:59:42 INFO network.SendingConnection: Initiating connection to [ip-10-80-91-13.ec2.internal/10.80.91.13:39149] 14/11/15 01:59:42 INFO network.SendingConnection: Connected to [ip-10-80-91-13.ec2.internal/10.80.91.13:39149], 1 messages pending 14/11/15 01:59:42 INFO network.ConnectionManager: Accepted connection from [ip-10-80-91-13.ec2.internal/10.80.91.13:56700] 14/11/15 01:59:42 WARN storage.BlockManager: Block input-0-1416016781800 already exists on this machine; not re-adding it 14/11/15 01:59:42 INFO receiver.BlockGenerator: Pushed block input-0-1416016781800 14/11/15 01:59:43 INFO storage.MemoryStore: ensureFreeSpace(256) called with curMem=3776, maxMem=938244833 14/11/15 01:59:43 INFO storage.MemoryStore: Block input-0-1416016782800 stored as values in memory (estimated size 256.0 B, free 894.8 MB) 14/11/15 01:59:43 INFO storage.BlockManagerInfo: Added input-0-1416016782800 in memory on ip-10-80-91-13.ec2.internal:39149 (size: 256.0 B, free: 894.8 MB) 14/11/15 01:59:43 INFO storage.BlockManagerMaster: Updated info of block input-0-1416016782800 14/11/15 01:59:43 WARN storage.BlockManager: Block input-0-1416016782800 already exists on this machine; not re-adding it 14/11/15 01:59:43 INFO receiver.BlockGenerator: Pushed block input-0-1416016782800 14/11/15 01:59:44 INFO scheduler.ReceiverTracker: Stream 0 received 2 blocks 14/11/15 01:59:44 INFO scheduler.ReceiverTracker: Stream 1 received 0 blocks 14/11/15 01:59:44 INFO scheduler.JobScheduler: Added jobs for time 1416016784000 ms 14/11/15 01:59:46 INFO scheduler.ReceiverTracker: Stream 0 received 0 blocks 14/11/15 01:59:46 INFO scheduler.ReceiverTracker: Stream 1 received 0 blocks 14/11/15 01:59:46 INFO scheduler.JobScheduler: Added jobs for time 1416016786000 ms 14/11/15 01:59:46 INFO impl.CWPublisherRunnable: Successfully published 17 datums. 14/11/15 01:59:46 INFO storage.MemoryStore: ensureFreeSpace(248) called with curMem=4032, maxMem=938244833 14/11/15 01:59:46 INFO storage.MemoryStore: Block input-1-1416016786000 stored as values in memory (estimated size 248.0 B, free 894.8 MB) 14/11/15 01:59:46 INFO storage.BlockManagerInfo: Added input-1-1416016786000 in memory on ip-10-80-91-13.ec2.internal:39149 (size: 248.0 B, free: 894.8 MB) 14/11/15 01:59:46 INFO storage.BlockManagerMaster: Updated info of block input-1-1416016786000 14/11/15 01:59:46 WARN storage.BlockManager: Block input-1-1416016786000 already exists on this machine; not re-adding it 14/11/15 01:59:46 INFO receiver.BlockGenerator: Pushed block input-1-1416016786000 14/11/15 01:59:46 INFO impl.CWPublisherRunnable: Successfully published 14 datums. 14/11/15 01:59:48 INFO scheduler.ReceiverTracker: Stream 0 received 0 blocks 14/11/15 01:59:48 INFO storage.MemoryStore: ensureFreeSpace(264) called with curMem=4280, maxMem=938244833 14/11/15 01:59:48 INFO scheduler.ReceiverTracker: Stream 1 received 1 blocks 14/11/15 01:59:48 INFO storage.MemoryStore: Block input-0-1416016787800 stored as values in memory (estimated size 264.0 B, free 894.8 MB) 14/11/15 01:59:48 INFO storage.BlockManagerInfo: Added input-0-1416016787800 in memory on ip-10-80-91-13.ec2.internal:39149 (size: 264.0 B, free: 894.8 MB) 14/11/15 01:59:48 INFO storage.BlockManagerMaster: Updated info of block input-0-1416016787800 14/11/15 01:59:48 INFO scheduler.JobScheduler: Added jobs for time 1416016788000 ms 14/11/15 01:59:48 WARN storage.BlockManager: Block input-0-1416016787800 already exists on this machine; not re-adding it 14/11/15 01:59:48 INFO receiver.BlockGenerator: Pushed block input-0-1416016787800 14/11/15 01:59:50 INFO scheduler.ReceiverTracker: Stream 0 received 1 blocks 14/11/15 01:59:50 INFO scheduler.ReceiverTracker: Stream 1 received 0 blocks 14/11/15 01:59:50 INFO scheduler.JobScheduler: Added jobs for time 1416016790000 ms 14/11/15 01:59:51 INFO storage.MemoryStore: ensureFreeSpace(264) called with curMem=4544, maxMem=938244833 14/11/15 01:59:51 INFO storage.MemoryStore: Block input-0-1416016790800 stored as values in memory (estimated size 264.0 B, free 894.8 MB) 14/11/15 01:59:51 INFO storage.BlockManagerInfo: Added input-0-1416016790800 in memory on ip-10-80-91-13.ec2.internal:39149 (size: 264.0 B, free: 894.8 MB) 14/11/15 01:59:51 INFO storage.BlockManagerMaster: Updated info of block input-0-1416016790800 14/11/15 01:59:51 WARN storage.BlockManager: Block input-0-1416016790800 already exists on this machine; not re-adding it 14/11/15 01:59:51 INFO receiver.BlockGenerator: Pushed block input-0-1416016790800

推荐答案

这可能是与你有多少工作线程了。我有同样的问题，当我跑的应用程序与本地--master [2]。我花了大量的时间寻找答案，并没有发现什么。只是出于好奇，我改成--master本地[4]和它的工作。我不知道的根本原因。也许有人更熟悉星火可以启发我们。

This might have something to do with how many worker thread you got. I had the same problem when I ran the app with --master local[2]. I spent numerous hours searching for an answer and found nothing. Just out of curiosity, I changed to --master local[4] and it worked. I do not know the root cause. Maybe somebody more familiar with Spark can enlighten us.

请注意：在我的情况，我室壁运动流有两个碎片。因此，应用程序创建了两个输入流，每一个碎片。

Note: in my case, my Kinesis stream had two shards. So the app created two input streams, one for each shard.

这篇关于阿帕奇星火室壁运动样品不工作的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！