标签:方式 选项 连接 本地 logfile ext ORC numa cache
spark框架是用scala写的,运行在Java虚拟机(JVM)上。支持Python、Java、Scala或R多种语言编写客户端应用。访问http://spark.apache.org/downloads.html选择预编译的版本进行下载。
打开终端,将工作路径转到下载的spark压缩包所在的目录,然后解压压缩包。
可使用如下命令:
cd ~
tar -xf spark-2.2.2-bin-hadoop2.7.tgz -C /opt/module/
cd spark-2.2.2-bin-hadoop2.7
ls
注:tar命令中x标记指定tar命令执行解压缩操作,f标记指定压缩包的文件名。
包含用来入门spark的简单使用说明
包含可用来和spark进行各种方式交互的一系列可执行文件
包含spark项目主要组件的源代码
包含一些可查看和运行的spark程序,对学习spark的API非常有帮助
./bin/run-example SparkPi 10
./bin/spark-shell --master local[2]
# --master选项指定运行模式。local是指使用一个线程本地运行;local[N]是指使用N个线程本地运行。
./bin/pyspark --master local[2]
./bin/sparkR --master local[2]
#支持多种语言提交
./bin/spark-submit examples/src/main/python/pi.py 10
./bin/spark-submit examples/src/main/r/dataframe.R
...
使用spark-shell脚本进行交互式分析。
scala> val textFile = spark.read.textFile("README.md")
textFile: org.apache.spark.sql.Dataset[String] = [value: string]
scala> textFile.count() // Number of items in this Dataset
res0: Long = 126 // May be different from yours as README.md will change over time, similar to other outputs
scala> textFile.first() // First item in this Dataset
res1: String = # Apache Spark
#使用filter算子返回原DataSet的子集
scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))
linesWithSpark: org.apache.spark.sql.Dataset[String] = [value: string]
#拉链方式
scala> textFile.filter(line => line.contains("Spark")).count() // How many lines contain "Spark"?
res3: Long = 15
#使用DataSet的转换和动作查找最多单词的行
scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)
res4: Long = 15
#统计单词个数
scala> val wordCounts = textFile.flatMap(line => line.split(" ")).groupByKey(identity).count()
wordCounts: org.apache.spark.sql.Dataset[(String, Long)] = [value: string, count(1): bigint]
scala> wordCounts.collect()
res6: Array[(String, Int)] = Array((means,1), (under,2), (this,3), (Because,1), (Python,2), (agree,1), (cluster.,1), ...)
使用pyspark脚本进行交互式分析
>>> textFile = spark.read.text("README.md")
>>> textFile.count() # Number of rows in this DataFrame
126
>>> textFile.first() # First row in this DataFrame
Row(value=u‘# Apache Spark‘)
#filter过滤
>>> linesWithSpark = textFile.filter(textFile.value.contains("Spark"))
#拉链方式
>>> textFile.filter(textFile.value.contains("Spark")).count() # How many lines contain "Spark"?
15
#查找最多单词的行
>>> from pyspark.sql.functions import *
>>> textFile.select(size(split(textFile.value, "\s+")).name("numWords")).agg(max(col("numWords"))).collect()
[Row(max(numWords)=15)]
#统计单词个数
>>> wordCounts = textFile.select(explode(split(textFile.value, "\s+")).alias("word")).groupBy("word").count()
>>> wordCounts.collect()
[Row(word=u‘online‘, count=1), Row(word=u‘graphs‘, count=1), ...]
spark除了交互式运行之外,spark也可以在Java、Scala或Python的独立程序中被连接使用。
独立应用与shell的主要区别在于需要自行初始化SparkContext。
分别统计包含单词a和单词b的行数
/* 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()
}
}
运行应用
# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit --class "SimpleApp" --master local[4] target/scala-2.11/simple-project_2.11-1.0.jar
...
Lines with a: 46, Lines with b: 23
分别统计包含单词a和单词b的行数
/* SimpleApp.java */
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
public class SimpleApp {
public static void main(String[] args) {
String logFile = "YOUR_SPARK_HOME/README.md"; // Should be some file on your system
SparkSession spark = SparkSession.builder().appName("Simple Application").getOrCreate();
Dataset<String> logData = spark.read().textFile(logFile).cache();
long numAs = logData.filter(s -> s.contains("a")).count();
long numBs = logData.filter(s -> s.contains("b")).count();
System.out.println("Lines with a: " + numAs + ", lines with b: " + numBs);
spark.stop();
}
}
运行应用
# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit --class "SimpleApp" --master local[4] target/simple-project-1.0.jar
...
Lines with a: 46, Lines with b: 23
分别统计包含单词a和单词b的行数
setup.py脚本添加内容
install_requires=[
‘pyspark=={site.SPARK_VERSION}‘
]
"""SimpleApp.py"""
from pyspark.sql import SparkSession
logFile = "YOUR_SPARK_HOME/README.md" # Should be some file on your system
spark = SparkSession.builder().appName(appName).master(master).getOrCreate()
logData = spark.read.text(logFile).cache()
numAs = logData.filter(logData.value.contains(‘a‘)).count()
numBs = logData.filter(logData.value.contains(‘b‘)).count()
print("Lines with a: %i, lines with b: %i" % (numAs, numBs))
spark.stop()
运行应用
# Use spark-submit to run your application
$ YOUR_SPARK_HOME/bin/spark-submit --master local[4] SimpleApp.py
...
Lines with a: 46, Lines with b: 23
标签:方式 选项 连接 本地 logfile ext ORC numa cache
原文地址:http://blog.51cto.com/12967015/2163611