1. Hadoop
scala
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.{IntWritable, Text}
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.Mapper
import org.apache.hadoop.mapreduce.Reducer
object WordCount {
class TokenizerMapper extends Mapper[Object, Text, Text, IntWritable] {
private final val one = new IntWritable(1)
private val word = new Text()
override def map(key: Object, value: Text, context: Context): Unit = {
value.toString.split("\\s+").foreach { token =>
word.set(token)
context.write(word, one)
}
}
}
class IntSumReducer extends Reducer[Text, IntWritable, Text, IntWritable] {
private val result = new IntWritable()
override def reduce(key: Text, values: java.lang.Iterable[IntWritable], context: Context): Unit = {
val sum = values.asScala.map(_.get).sum
result.set(sum)
context.write(key, result)
}
}
def main(args: Array[String]): Unit = {
val conf = new Configuration()
val job = Job.getInstance(conf, "word count")
job.setJarByClass(WordCount.getClass)
job.setMapperClass(classOf[TokenizerMapper])
job.setCombinerClass(classOf[IntSumReducer])
job.setReducerClass(classOf[IntSumReducer])
job.setOutputKeyClass(classOf[Text])
job.setOutputValueClass(classOf[IntWritable])
val outputPath = new Path(args(1))
outputPath.getFileSystem(conf).delete(outputPath, true)
job.waitForCompletion(true)
}
}
2. Apache Spark
scala
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
object SQLExample {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("SQLExample").setMaster("local")
val spark = SparkSession.builder().config(conf).getOrCreate()
val df = spark.read.json("input.json")
df.createOrReplaceTempView("people")
val result = spark.sql("SELECT name, age FROM people WHERE age >= 18")
result.show()
spark.stop()
}
}
