rssanders3

cd /opt
sudo wget --no-check-certificate https://www.python.org/ftp/python/2.7.9/Python-2.7.9.tar.xz
tar xf Python-2.7.9.tar.xz
cd Python-2.7.9
./configure --prefix=/usr/local
make && make altinstall

ls -ltr /usr/local/bin/python*

vi ~/.bashrc

rssanders3

val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(2))

// Create direct kafka stream with brokers and topics
val topicsSet = topics.split(",").toSet
val kafkaParams = Map[String, Object](
  ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
  ConsumerConfig.GROUP_ID_CONFIG -> groupId,
  ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
  ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer])

rssanders3

var TRIGGER_STOP = false
var ssc: StreamingContext = …

// Define Stream Creation, Transformations and Actions here.

ssc.start()

var isStopped = false
while (!isStopped) {
     isStopped = ssc.awaitTerminationOrTimeout(SPARK_SHUTDOWN_CHECK_MILLIS)

rssanders3

val checkpointDirectory = "hdfs://..."   // define checkpoint directory

// Function to create and setup a new StreamingContext
def functionToCreateContext(): StreamingContext = {
  val ssc = new StreamingContext(...)   // new context
  val lines = ssc.socketTextStream(...) // create DStreams
  ...
  ssc.checkpoint(checkpointDirectory)   // set checkpoint directory
  ssc  // Return the StreamingContext
}

rssanders3

val sparkConf = new SparkConf()
  .setAppName("App")
  .set("spark.yarn.maxAppAttempts", "1")
  .set("spark.yarn.am.attemptFailuresValidityInterval", "2h")

val ssc = new StreamingContext(sparkConf, Seconds(2))

rssanders3

val storedOffsets: Option[mutable.Map[TopicPartition, Long]] = loadOffsets(spark, kuduContext)

val kafkaDStream = storedOffsets match {
    case None =>
        LOGGER.info("storedOffsets was None")
        kafkaParams += ("auto.offset.reset" -> "latest")
        KafkaUtils.createDirectStream[String, Array[Byte]]
            (ssc, PreferConsistent, ConsumerStrategies.Subscribe[String, Array[Byte]]
                (topicsSet, kafkaParams)
        )

rssanders3

// Cell 1
sqlContext

// Cell 2
sqlContext.sql("show tables").collect()
val got = sqlContext.sql("select * from got")

// Cell 3
got.limit(10).collect()

rssanders3

import org.apache.spark.sql.functions._

val allegiancesCleanupUDF = udf[String, String] (_.toLowerCase().replace("house ", ""))
val isDeathUDF = udf{ deathYear: Integer => if(deathYear != null) 1 else 0}

val gotCleaned = got.filter("Allegiances != \"None\"").withColumn("Allegiances", allegiancesCleanupUDF($"Allegiances")).withColumn("isDeath", isDeathUDF($"Death Year"))

display(gotCleaned)

rssanders3

// Cell 1
sc

// Cell 2
sc.parallelize(1 to 5).collect()

rssanders3

import site
import os
SITE_PACKAGES = site.getsitepackages()
print "All Site Packages: " + str(SITE_PACKAGES)
for site_package in SITE_PACKAGES:
  test_path = site_package + "/airflow"
  if os.path.exists(test_path):
    AIRFLOW_INSTALL_DIR = test_path
print "Site Page Containing Airflow: " + str(AIRFLOW_INSTALL_DIR)