gistfile1.txt

  val sc = new SparkContext(conf)
  val csc = new CassandraSQLContext(sc)
  CassandraConnector(conf).withSessionDo { session =>
    session.execute(s"CREATE KEYSPACE IF NOT EXISTS adlogs " +
      s"WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 }")
	session.execute(s"CREATE TABLE IF NOT EXISTS adlogs.adimpressions " +
      s"(timestamp bigint, publisher text, advertiser text, " +
      "website text, geo text, bid double, cookie text, primary key (timestamp, cookie))")
  }
 
 
  import com.datastax.spark.connector.streaming._
 
  kafkaStream.map(_._2).map(ad =>
    (ad.getTimestamp, ad.getPublisher, ad.getAdvertiser, ad.getWebsite,
     ad.getGeo, ad.getBid, ad.getCookie))
    .saveToCassandra("adlogs", "adimpressions",
      SomeColumns("timestamp", "publisher", "advertiser", "website", "geo", "bid", "cookie"))