{"action":"opened","issue":{"url":"https://api.github.com/repos/divolte/divolte-examples/issues/4","labels_url":"https://api.github.com/repos/divolte/divolte-examples/issues/4/labels{/name}","comments_url":"https://api.github.com/repos/divolte/divolte-examples/issues/4/comments","events_url":"https://api.github.com/repos/divolte/divolte-examples/issues/4/events","html_url":"https://github.com/divolte/divolte-examples/issues/4","id":87291326,"number":4,"title":"SparkStreaming and Kafka","user":{"login":"BHMath","id":6143089,"avatar_url":"https://avatars.githubusercontent.com/u/6143089?v=3","gravatar_id":"","url":"https://api.github.com/users/BHMath","html_url":"https://github.com/BHMath","followers_url":"https://api.github.com/users/BHMath/followers","following_url":"https://api.github.com/users/BHMath/following{/other_user}","gists_url":"https://api.github.com/users/BHMath/gists{/gist_id}","starred_url":"https://api.github.com/users/BHMath/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/BHMath/subscriptions","organizations_url":"https://api.github.com/users/BHMath/orgs","repos_url":"https://api.github.com/users/BHMath/repos","events_url":"https://api.github.com/users/BHMath/events{/privacy}","received_events_url":"https://api.github.com/users/BHMath/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":0,"created_at":"2015-06-11T10:12:32Z","updated_at":"2015-06-11T10:12:32Z","closed_at":null,"body":"I'm trying to implement your example to stream my divolte's data from kafka but I can't make it work.\r\n\r\nI created a maven project with all libraries needed.\r\n\r\nI create a scala object exactly like https://github.com/divolte/divolte-examples/blob/master/spark/src/main/scala/io/divolte/examples/DivolteSparkStreamingExample.scala#L84\r\nI define my spark context like this\r\n```\r\n// Kafka configuration.\r\n  val consumerConfig = Map(\r\n    \"group.id\" -> \"divolte-spark-streaming-cassandra\",\r\n    \"zookeeper.connect\" -> \"127.0.0.1:2181\",\r\n    \"auto.commit.interval.ms\" -> \"5000\",\r\n    \"auto.offset.reset\" -> \"largest\")\r\n  val topicSettings = Map(\"divolte\" -> Runtime.getRuntime.availableProcessors())\r\n\r\n  val conf = new SparkConf().setAppName(\"DKCtC\").setMaster(\"local\")\r\n\r\n  // Create the spark and streaming contexts. Micro batches per second.\r\n  val sc = new SparkContext(conf)\r\n  val ssc = new StreamingContext(sc, Seconds(3))\r\n```\r\nmy divolte and kafka is running inside a VM with all port forwarded and I run my scala from eclipse.\r\nBut every time something arrives in kafka my program seems to stop and I can't visualize any data.\r\n\r\nHere my full scala file\r\n```\r\npackage io.divolte\r\n\r\n/**\r\n * @author A526839\r\n */\r\nimport java.io.File\r\nimport java.nio.file.Files\r\nimport java.nio.file.Paths\r\n\r\nimport io.divolte.spark.avro._\r\nimport org.apache.avro.generic.GenericRecord\r\nimport org.apache.spark.SparkConf\r\nimport org.apache.spark.SparkContext\r\nimport org.apache.spark.storage.StorageLevel\r\nimport org.apache.spark.streaming.dstream.DStream\r\nimport org.apache.spark.streaming.{ Minutes, Seconds, StreamingContext }\r\n\r\nobject DivolteSparkStreamingToCassandra extends App {\r\n  // Local only for webutils\r\n  val sep = File.separator\r\n  val current = new File(\".\").getAbsolutePath.replace(s\"$sep.\", \"\")\r\n  System.setProperty(\"hadoop.home.dir\", s\"$current${sep}src${sep}main${sep}resources${sep}\")\r\n\r\n  // Kafka configuration.\r\n  val consumerConfig = Map(\r\n    \"group.id\" -> \"divolte-spark-streaming-cassandra\",\r\n    \"zookeeper.connect\" -> \"127.0.0.1:2181\",\r\n    \"auto.commit.interval.ms\" -> \"5000\",\r\n    \"auto.offset.reset\" -> \"largest\")\r\n  val topicSettings = Map(\"divolte\" -> Runtime.getRuntime.availableProcessors())\r\n\r\n  val conf = new SparkConf().setAppName(\"DKCtC\").setMaster(\"local\")\r\n\r\n  // Create the spark and streaming contexts. Micro batches per second.\r\n  // For local\r\n  val sc = new SparkContext(conf)\r\n  //  val sc = new SparkContext()\r\n  val ssc = new StreamingContext(sc, Seconds(1))\r\n\r\n  //   Where checkpoints will be stored.\r\n  val checkPointDirectory = Files.createTempDirectory(\"divolte-collector-streaming-cassandra\")\r\n  checkPointDirectory.toFile.deleteOnExit()\r\n  ssc.checkpoint(checkPointDirectory.toString)\r\n\r\n  // Establish the source event stream.\r\n  val stream = ssc.divolteStream[GenericRecord](consumerConfig, topicSettings, StorageLevel.MEMORY_ONLY)\r\n\r\n  var passes = 0\r\n  \r\n  val events = stream.toRecords\r\n  \r\n  events.foreachRDD { event =>\r\n    println(s\"passes: ${passes}\")\r\n    passes = passes + 1\r\n  }\r\n\r\n  // Assuming sessions last half an hour, create a stream counting the number of active users.\r\n  val partyIdStream: DStream[String] = stream.mapValues(_ => None).map(_._1)\r\n  val uniquePartyCountStream = partyIdStream\r\n    .countByValueAndWindow(windowDuration = Minutes(30), slideDuration = Seconds(5))\r\n    .map(_._1)\r\n    .count()\r\n  \r\n    // Create a stream creating just the page locations for each event.\r\n    val pageIdStream: DStream[String] = stream\r\n      .fields(\"location\")\r\n      .flatMap(_._2.head.map(_.asInstanceOf[String]))\r\n  \r\n    // Create streams counting the top 5 locations during over various intervals.\r\n    val locationCountsStreams = Seq(1L, 5L, 15L).map { windowMinutes =>\r\n      val window = Minutes(windowMinutes)\r\n      window ->\r\n        pageIdStream.countByValueAndWindow(windowDuration = window, slideDuration = Seconds(15))\r\n    }\r\n\r\n  // Configure our displays:\r\n  //   - Every URL, as they occur.\r\n  //   - The active count.\r\n  //   - The top 5 counts.\r\n  uniquePartyCountStream.foreachRDD { rdd =>\r\n    println(s\"Active user count: ${rdd.collect().head}\")\r\n  }\r\n    pageIdStream.foreachRDD { rdd =>\r\n      rdd.collect().foreach { location =>\r\n        println(s\"Location: $location\")\r\n      }\r\n    }\r\n    locationCountsStreams.foreach {\r\n      case (window, locationCountsStream) =>\r\n        locationCountsStream.foreachRDD { locationCounts =>\r\n          val top5 = locationCounts.top(5)(Ordering.by(_._2))\r\n          println(s\"=== Top 5 (${window.prettyPrint}) ===\")\r\n          top5.zipWithIndex.foreach {\r\n            case ((location, count), index) =>\r\n              println(s\" ${index + 1}. $location ($count)\")\r\n          }\r\n        }\r\n    }\r\n\r\n  // Start the stream and await termination.\r\n  ssc.start()\r\n  ssc.awaitTermination()\r\n}\r\n```\r\n\r\nAnd here the result with log4j configure to only display warn\r\n```\r\n15/06/11 12:07:12 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\r\n15/06/11 12:07:14 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.\r\npasses: 0\r\n15/06/11 12:07:16 WARN RangeAssignor: No broker partitions consumed by consumer thread divolte-spark-streaming-cassandra_EFR00273-1434017235784-151ccd10-1 for topic divolte\r\n15/06/11 12:07:16 WARN RangeAssignor: No broker partitions consumed by consumer thread divolte-spark-streaming-cassandra_EFR00273-1434017235784-151ccd10-3 for topic divolte\r\n15/06/11 12:07:16 WARN RangeAssignor: No broker partitions consumed by consumer thread divolte-spark-streaming-cassandra_EFR00273-1434017235784-151ccd10-2 for topic divolte\r\npasses: 1\r\npasses: 2\r\npasses: 3\r\npasses: 4\r\n```\r\nat my fourth pass I refreshed my website then an event have been set to kafka I can see it with\r\n```\r\n[www@localhost kafka]$ ./bin/kafka-console-consumer.sh --zookeeper 127.0.0.1:2181 --topic divolte --from-beginning\r\nChromeGoogle Inc.Browser43.0.2357.124\"Personal computerWindows6.1,Microsoft Corporation.displayPage\r\n```\r\n\r\nCan you help me to read my data from kafka with a spark program. My goal is to store this data inside a Cassandra db.\r\n\r\nThanks ahead\r\n"}}