Estoy intentando con IntelliJ y Spark Streaming conectar con la API de Twitter para poder acceder a los tweets. Por el momento lo único que quiero que haga es que imprima cada tweet por la consola.
Parece que viendo los errores, el problema viene antes del SparkContext, ya que veo que el tiempo de 4 segundos que le he puesto si que lo está haciendo. No sé si el problema viene en el "ConfigurationBuilder" que hay algo que estoy haciendo mal.
El código que tengo actualmente es el siguiente:
import twitter4j.conf.ConfigurationBuilder
import twitter4j.auth.OAuthAuthorization
import org.apache.spark.streaming.twitter._
import org.apache.spark._
import org.apache.spark.streaming._
object Main {
def main(args: Array[String]): Unit = {
val apiKey = "XXXX"
val apiKeySecret = "XXXX"
val accessToken = "XXXX"
val accessTokenSecret = "XXXX"
val cb = new ConfigurationBuilder
cb.setDebugEnabled(true)
.setOAuthConsumerKey(apiKey)
.setOAuthConsumerSecret(apiKeySecret)
.setOAuthAccessToken(accessToken)
.setOAuthAccessTokenSecret(accessTokenSecret)
val conf = new SparkConf().setAppName("twitter_spark_streaming").setMaster("local[*]")
val ssc = new StreamingContext(conf, Seconds(4))
val auth = new OAuthAuthorization(cb.build)
//val tweets = TwitterUtils.createStream(ssc,Some(auth))
val tweets = TwitterUtils.createStream(ssc, Some(auth), null, StorageLevel.MEMORY_AND_DISK_2)
val statuses = tweets.map(status => status.getText())
statuses.print()
ssc.start()
ssc.awaitTermination()
}
}
Me está dando el siguiente error y no consigo encontrar la solución.
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
19/08/09 13:33:20 INFO SparkContext: Running Spark version 2.3.0
19/08/09 13:33:20 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
19/08/09 13:33:20 INFO SparkContext: Submitted application: twitter_spark_streaming
19/08/09 13:33:20 INFO SecurityManager: Changing view acls to: aresa
19/08/09 13:33:20 INFO SecurityManager: Changing modify acls to: aresa
19/08/09 13:33:20 INFO SecurityManager: Changing view acls groups to:
19/08/09 13:33:20 INFO SecurityManager: Changing modify acls groups to:
19/08/09 13:33:20 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(aresa); groups with view permissions: Set(); users with modify permissions: Set(aresa); groups with modify permissions: Set()
19/08/09 13:33:21 INFO Utils: Successfully started service 'sparkDriver' on port 54295.
19/08/09 13:33:21 INFO SparkEnv: Registering MapOutputTracker
19/08/09 13:33:21 INFO SparkEnv: Registering BlockManagerMaster
19/08/09 13:33:21 INFO BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
19/08/09 13:33:21 INFO BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
19/08/09 13:33:21 INFO DiskBlockManager: Created local directory at C:\Users\aresa\AppData\Local\Temp\blockmgr-89c10385-b401-4be9-bb71-c2868b0cb6f5
19/08/09 13:33:21 INFO MemoryStore: MemoryStore started with capacity 1988.7 MB
19/08/09 13:33:21 INFO SparkEnv: Registering OutputCommitCoordinator
19/08/09 13:33:21 INFO Utils: Successfully started service 'SparkUI' on port 4040.
19/08/09 13:33:21 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at http://MSI:4040
19/08/09 13:33:21 INFO Executor: Starting executor ID driver on host localhost
19/08/09 13:33:21 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 54336.
19/08/09 13:33:21 INFO NettyBlockTransferService: Server created on MSI:54336
19/08/09 13:33:21 INFO BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
19/08/09 13:33:21 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(driver, MSI, 54336, None)
19/08/09 13:33:21 INFO BlockManagerMasterEndpoint: Registering block manager MSI:54336 with 1988.7 MB RAM, BlockManagerId(driver, MSI, 54336, None)
19/08/09 13:33:21 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(driver, MSI, 54336, None)
19/08/09 13:33:21 INFO BlockManager: Initialized BlockManager: BlockManagerId(driver, MSI, 54336, None)
19/08/09 13:33:22 INFO ReceiverTracker: Starting 1 receivers
19/08/09 13:33:22 INFO ReceiverTracker: ReceiverTracker started
19/08/09 13:33:22 INFO TwitterInputDStream: Slide time = 4000 ms
19/08/09 13:33:22 INFO TwitterInputDStream: Storage level = Serialized 1x Replicated
19/08/09 13:33:22 INFO TwitterInputDStream: Checkpoint interval = null
19/08/09 13:33:22 INFO TwitterInputDStream: Remember interval = 4000 ms
19/08/09 13:33:22 INFO TwitterInputDStream: Initialized and validated org.apache.spark.streaming.twitter.TwitterInputDStream@575b0ffe
19/08/09 13:33:22 INFO MappedDStream: Slide time = 4000 ms
19/08/09 13:33:22 INFO MappedDStream: Storage level = Serialized 1x Replicated
19/08/09 13:33:22 INFO MappedDStream: Checkpoint interval = null
19/08/09 13:33:22 INFO MappedDStream: Remember interval = 4000 ms
19/08/09 13:33:22 INFO MappedDStream: Initialized and validated org.apache.spark.streaming.dstream.MappedDStream@45d09d8b
19/08/09 13:33:22 INFO ForEachDStream: Slide time = 4000 ms
19/08/09 13:33:22 INFO ForEachDStream: Storage level = Serialized 1x Replicated
19/08/09 13:33:22 INFO ForEachDStream: Checkpoint interval = null
19/08/09 13:33:22 INFO ForEachDStream: Remember interval = 4000 ms
19/08/09 13:33:22 INFO ForEachDStream: Initialized and validated org.apache.spark.streaming.dstream.ForEachDStream@1fbf4d6a
19/08/09 13:33:22 INFO ReceiverTracker: Receiver 0 started
19/08/09 13:33:22 INFO RecurringTimer: Started timer for JobGenerator at time 1565350404000
19/08/09 13:33:22 INFO JobGenerator: Started JobGenerator at 1565350404000 ms
19/08/09 13:33:22 INFO JobScheduler: Started JobScheduler
19/08/09 13:33:22 INFO DAGScheduler: Got job 0 (start at Main.scala:50) with 1 output partitions
19/08/09 13:33:22 INFO DAGScheduler: Final stage: ResultStage 0 (start at Main.scala:50)
19/08/09 13:33:22 INFO DAGScheduler: Parents of final stage: List()
19/08/09 13:33:22 INFO DAGScheduler: Missing parents: List()
19/08/09 13:33:22 INFO StreamingContext: StreamingContext started
19/08/09 13:33:22 INFO DAGScheduler: Submitting ResultStage 0 (Receiver 0 ParallelCollectionRDD[0] at makeRDD at ReceiverTracker.scala:613), which has no missing parents
19/08/09 13:33:22 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 46.5 KB, free 1988.7 MB)
19/08/09 13:33:22 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 15.8 KB, free 1988.6 MB)
19/08/09 13:33:22 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on MSI:54336 (size: 15.8 KB, free: 1988.7 MB)
19/08/09 13:33:22 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1039
19/08/09 13:33:22 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 0 (Receiver 0 ParallelCollectionRDD[0] at makeRDD at ReceiverTracker.scala:613) (first 15 tasks are for partitions Vector(0))
19/08/09 13:33:22 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
19/08/09 13:33:22 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, executor driver, partition 0, PROCESS_LOCAL, 10840 bytes)
19/08/09 13:33:22 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
19/08/09 13:33:23 INFO RecurringTimer: Started timer for BlockGenerator at time 1565350403200
19/08/09 13:33:23 INFO BlockGenerator: Started BlockGenerator
19/08/09 13:33:23 INFO BlockGenerator: Started block pushing thread
19/08/09 13:33:23 INFO ReceiverTracker: Registered receiver for stream 0 from MSI:54295
19/08/09 13:33:23 INFO ReceiverSupervisorImpl: Starting receiver 0
19/08/09 13:33:23 INFO ReceiverSupervisorImpl: Called receiver 0 onStart
19/08/09 13:33:23 INFO ReceiverSupervisorImpl: Waiting for receiver to be stopped
19/08/09 13:33:23 WARN ReceiverSupervisorImpl: Restarting receiver with delay 2000 ms: Error starting Twitter stream
java.lang.NullPointerException
at org.apache.spark.streaming.twitter.TwitterReceiver.onStart(TwitterInputDStream.scala:89)
at org.apache.spark.streaming.receiver.ReceiverSupervisor.startReceiver(ReceiverSupervisor.scala:149)
at org.apache.spark.streaming.receiver.ReceiverSupervisor.start(ReceiverSupervisor.scala:131)
at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:600)
at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:590)
at org.apache.spark.SparkContext$$anonfun$34.apply(SparkContext.scala:2178)
at org.apache.spark.SparkContext$$anonfun$34.apply(SparkContext.scala:2178)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
19/08/09 13:33:23 INFO ReceiverSupervisorImpl: Stopping receiver with message: Restarting receiver with delay 2000ms: Error starting Twitter stream: java.lang.NullPointerException
Exception in thread "receiver-supervisor-future-0" java.lang.AbstractMethodError
at org.apache.spark.internal.Logging$class.initializeLogIfNecessary(Logging.scala:99)
at org.apache.spark.streaming.twitter.TwitterReceiver.initializeLogIfNecessary(TwitterInputDStream.scala:60)
at org.apache.spark.internal.Logging$class.log(Logging.scala:46)
at org.apache.spark.streaming.twitter.TwitterReceiver.log(TwitterInputDStream.scala:60)
at org.apache.spark.internal.Logging$class.logInfo(Logging.scala:54)
at org.apache.spark.streaming.twitter.TwitterReceiver.logInfo(TwitterInputDStream.scala:60)
at org.apache.spark.streaming.twitter.TwitterReceiver.onStop(TwitterInputDStream.scala:106)
at org.apache.spark.streaming.receiver.ReceiverSupervisor.stopReceiver(ReceiverSupervisor.scala:170)
at org.apache.spark.streaming.receiver.ReceiverSupervisor$$anonfun$restartReceiver$1.apply$mcV$sp(ReceiverSupervisor.scala:194)
at org.apache.spark.streaming.receiver.ReceiverSupervisor$$anonfun$restartReceiver$1.apply(ReceiverSupervisor.scala:189)
at org.apache.spark.streaming.receiver.ReceiverSupervisor$$anonfun$restartReceiver$1.apply(ReceiverSupervisor.scala:189)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
19/08/09 13:33:24 INFO JobScheduler: Added jobs for time 1565350404000 ms
19/08/09 13:33:24 INFO JobScheduler: Starting job streaming job 1565350404000 ms.0 from job set of time 1565350404000 ms
19/08/09 13:33:24 INFO JobScheduler: Added jobs for time 1565350404000 ms
19/08/09 13:33:24 INFO JobScheduler: Starting job streaming job 1565350404000 ms.0 from job set of time 1565350404000 ms
-------------------------------------------
Time: 1565350404000 ms
-------------------------------------------
19/08/09 13:33:24 INFO JobScheduler: Finished job streaming job 1565350404000 ms.0 from job set of time 1565350404000 ms
19/08/09 13:33:24 INFO JobScheduler: Total delay: 0,043 s for time 1565350404000 ms (execution: 0,005 s)
19/08/09 13:33:24 INFO ReceivedBlockTracker: Deleting batches:
19/08/09 13:33:24 INFO InputInfoTracker: remove old batch metadata:
Process finished with exit code -1
No sé si el problema puede venir por incompatibilidades de estas dependencias.
Muchas gracias!