I'm trying to run the word count example described here, but the DStream reading from the Kinesis stream is always empty.
This is how I'm running:
- Launched an AWS EMR cluster in version 6.5.0 (Running spark 3.1.2)
- SSHed into the master instance
- ran:
spark-example --packages org.apache.spark:spark-streaming-kinesis-asl_2.12:3.1.2 streaming.JavaKinesisWordCountASL streaming_test streaming_test https://kinesis.sa-east-1.amazonaws.com - In another tab, ran:
spark-example --packages org.apache.spark:spark-streaming-kinesis-asl_2.12:3.1.2 streaming.KinesisWordProducerASL streaming-test https://kinesis.sa-east-1.amazonaws.com 100 10
Additional info:
- EMR cluster with 2 m5.xlarge instances
- Kinesis with a single shard only
- I can fetch records from the stream using boto3
- A DynamoDB table was indeed created for storing checkpoints, but nothing was written on it
Logs (This is just a sample - after it finishes initializing, it keeps repeating that pattern of pprint with no records, followed by a bunch of spark related logs, then followed again by another pprint with no records)
GiB)
22/01/27 21:39:46 INFO TaskSetManager: Starting task 1.0 in stage 8.0 (TID 77) (ip-10-0-13-187.sa-east-1.compute.internal, executor 1, partition 6, PROCESS_LOCAL, 4443 bytes) taskResourceAssignments Map()
22/01/27 21:39:46 INFO TaskSetManager: Finished task 0.0 in stage 8.0 (TID 76) in 19 ms on ip-10-0-13-187.sa-east-1.compute.internal (executor 1) (1/3)
22/01/27 21:39:46 INFO TaskSetManager: Starting task 2.0 in stage 8.0 (TID 78) (ip-10-0-13-187.sa-east-1.compute.internal, executor 1, partition 7, PROCESS_LOCAL, 4443 bytes) taskResourceAssignments Map()
22/01/27 21:39:46 INFO TaskSetManager: Finished task 1.0 in stage 8.0 (TID 77) in 10 ms on ip-10-0-13-187.sa-east-1.compute.internal (executor 1) (2/3)
22/01/27 21:39:46 INFO TaskSetManager: Finished task 2.0 in stage 8.0 (TID 78) in 8 ms on ip-10-0-13-187.sa-east-1.compute.internal (executor 1) (3/3)
22/01/27 21:39:46 INFO YarnScheduler: Removed TaskSet 8.0, whose tasks have all completed, from pool
22/01/27 21:39:46 INFO DAGScheduler: ResultStage 8 (print at JavaKinesisWordCountASL.java:190) finished in 0,042 s
22/01/27 21:39:46 INFO DAGScheduler: Job 4 is finished. Cancelling potential speculative or zombie tasks for this job
22/01/27 21:39:46 INFO YarnScheduler: Killing all running tasks in stage 8: Stage finished
22/01/27 21:39:46 INFO DAGScheduler: Job 4 finished: print at JavaKinesisWordCountASL.java:190, took 0,048372 s
-------------------------------------------
Time: 1643319586000 ms
-------------------------------------------
22/01/27 21:39:46 INFO JobScheduler: Finished job streaming job 1643319586000 ms.0 from job set of time 1643319586000 ms
22/01/27 21:39:46 INFO JobScheduler: Total delay: 0,271 s for time 1643319586000 ms (execution: 0,227 s)
22/01/27 21:39:46 INFO ReceivedBlockTracker: Deleting batches:
Also, apparently, the Library does manage to connect to the Kinesis stream:
22/01/27 21:39:44 INFO KinesisInputDStream: Slide time = 2000 ms
22/01/27 21:39:44 INFO KinesisInputDStream: Storage level = Serialized 1x Replicated
22/01/27 21:39:44 INFO KinesisInputDStream: Checkpoint interval = null
22/01/27 21:39:44 INFO KinesisInputDStream: Remember interval = 2000 ms
22/01/27 21:39:44 INFO KinesisInputDStream: Initialized and validated org.apache.spark.streaming.kinesis.KinesisInputDStream@7cc3580b
Help would be very appreciated!