Unit 1: Big data tools and analytics practice

• Web UIs: NameNode on port 9870, ResourceManager on port 8088.

bashhdfs dfs -mkdir -p /lab/input
hdfs dfs -put retail.csv /lab/input/
hdfs dfs -ls /lab/input
hdfs dfs -cat /lab/input/retail.csv | head
hdfs dfs -get /lab/output/part-00000 result.txt
hdfs fsck /lab/input/retail.csv -files -blocks      # see blocks and replicas

bashhadoop jar $HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-*.jar wordcount /lab/input /lab/wc_out
hdfs dfs -cat /lab/wc_out/part-r-00000 | sort -k2 -nr | head

• Also run with Hadoop Streaming using Python mapper and reducer scripts.

sqlCREATE EXTERNAL TABLE retail (invoice STRING, item STRING, qty INT, price DOUBLE, country STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION '/lab/input' TBLPROPERTIES ("skip.header.line.count"="1");
SELECT country, ROUND(SUM(qty * price), 2) AS revenue FROM retail GROUP BY country ORDER BY revenue DESC LIMIT 10;

pythonfrom pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
spark = SparkSession.builder.appName("lab").getOrCreate()
df = spark.read.csv("hdfs:///lab/input/customers.csv", header=True, inferSchema=True)
v = VectorAssembler(inputCols=["recency", "frequency", "monetary"], outputCol="raw").transform(df)
v = StandardScaler(inputCol="raw", outputCol="features").fit(v).transform(v)
for k in range(2, 7):
    m = KMeans(k=k, seed=1).fit(v)
    print(k, ClusteringEvaluator().evaluate(m.transform(v)))     # silhouette

pythonfrom pyspark.ml.classification import DecisionTreeClassifier, NaiveBayes
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import StringIndexer
data = spark.read.csv("hdfs:///lab/input/bank.csv", header=True, inferSchema=True)
data = StringIndexer(inputCol="y", outputCol="label").fit(data).transform(data)
data = VectorAssembler(inputCols=["age", "balance", "duration", "campaign"], outputCol="features").transform(data)
train, test = data.randomSplit([0.8, 0.2], seed=7)
ev = MulticlassClassificationEvaluator(metricName="accuracy")
for clf in (DecisionTreeClassifier(maxDepth=5), NaiveBayes()):
    print(type(clf).__name__, ev.evaluate(clf.fit(train).transform(test)))

pythonfrom pyspark.ml.fpm import FPGrowth
from pyspark.sql.functions import collect_set
baskets = spark.read.csv("hdfs:///lab/input/retail.csv", header=True).groupBy("invoice").agg(collect_set("item").alias("items"))
fp = FPGrowth(itemsCol="items", minSupport=0.02, minConfidence=0.3).fit(baskets)
fp.freqItemsets.orderBy("freq", ascending=False).show(10)
fp.associationRules.orderBy("lift", ascending=False).show(10)      # antecedent, consequent, confidence, lift

• FP-Growth avoids Apriori's repeated candidate generation and scans, so it scales better on large data.