在spark2.0版本以前,spakr编程接口是RDD(Resilient Distributed Dataset,弹性分布式数据集),spark2.0版本即以上,RDD被Dataset取代,Dataset比RDD更为强大,在底层得到了许多优化了。当然2.0+版本仍然支持RDD,但官方建议使用Dataset。
spark的安全模式默认是关闭的,这意味着你可能收到攻击。
在spark的安装根目录下启动。
启动
./bin/spark-shell
读取一个文件用来创建一个新的数据集Dataset
对数据集进行操作
textFile.count()
textFile.first()
val linesWithSpark = textFile.filter(line => line.contains("Spark"))
textFile.filter(line => line.contains("Spark")).count()
启动
./bin/pyspark
textFile = spark.read.text("README.md")
textFile.count()
textFile.first()
linesWithSpark = textFile.filter(textFile.value.contains("Spark"))
textFile.filter(textFile.value.contains("Spark")).count()
1.查找文件中长度最大的字符串,并返回长度
textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
2.实现wordcounts
val wordCounts = textFile.flatMap(line => line.split(" ")).groupByKey(identity).count()
wordCounts.collect()

linesWithSpark.cache()
linesWithSpark.count()
/* SimpleApp.scala */
import org.apache.spark.sql.SparkSession
object SimpleApp {
def main(args: Array[String]) {
val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
val spark = SparkSession.builder.appName("Simple Application").getOrCreate()
val logData = spark.read.textFile(logFile).cache()
val numAs = logData.filter(line => line.contains("a")).count()
val numBs = logData.filter(line => line.contains("b")).count()
println(s"Lines with a: $numAs, Lines with b: $numBs")
spark.stop()
}
}
原文:https://www.cnblogs.com/twodoge/p/10741446.html