rssanders3

[core]

# Whether to load the examples that ship with Airflow. It's good to
# get started, but you probably want to set this to False in a production
# environment
load_examples = False

rssanders3

[core]

# Are DAGs paused by default at creation
dags_are_paused_at_creation = True

rssanders3

[core]

# The SqlAlchemy connection string to the metadata database.
# SqlAlchemy supports many different database engine, more information
# their website
sql_alchemy_conn = mysql://{USERNAME}:{PASSWORD}@{MYSQL_HOST}:3306/airflow

rssanders3

[core]

# The executor class that airflow should use. Choices include
# SequentialExecutor, LocalExecutor, CeleryExecutor
executor = CeleryExecutor

rssanders3

pip install apache-airflow=={AIRFLOW_VERSION}
pip install apache-airflow[hive]=={AIRFLOW_VERSION}
pip install apache-airflow[celery]=={AIRFLOW_VERSION}

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)

rssanders3

// Cell 1
sc

// Cell 2
sc.parallelize(1 to 5).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
sqlContext

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

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

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)
        )