woot.scala

  def getMultiSegmentRdd(
    sc: SparkContext,
    sqlContext: CassandraSQLContext,
    keyspace: String,
    table: String,
    tenantId: Int,
    segments: Array[String],
    columns: Array[String] = Array()) :
      SchemaRDD = {
    val source = sc.parallelize(segments).map(Tuple2(tennantId, _))
    val repart = source.repartitionByCassandraReplica(keyspace, table, sparkPartitions)

    var joinRdd = repart.joinWithCassandraTable[CassandraSQLRow](keyspace, table)

    if (segments.length != 0) {
      joinRdd = joinRdd.select(columns.map{c => new ColumnName(c)}:_*)
    }

    val selectedColumnNames = joinRdd.selectedColumnNames
    val tableDef = joinRdd.tableDef.columnByName

    val schema = StructType(
      selectedColumnNames.map {
        name => StructField(
          name,
          ColumnDataType.catalystDataType(tableDef(name).columnType, true),
          true)
      }
    )
    return sqlContext.applySchema(joinRdd.map{x => x._2}, schema)
  }