Spark

╰半橙微兮° 2022-07-16 04:15 119阅读 0赞

http://www.infoq.com/cn/articles/apache-spark-introduction/ 学习  
http://www.infoq.com/cn/articles/apache-spark-streaming shu  
http://www.open-open.com/lib/view/open1432192407317.html  
    安装scala,下载scala-2.11.8.tgz，解压到/opt/scala-2.11.8  
           下地地址： http://www.scala-lang.org/  
    安装Spark，下载spark-1.3.1-bin-hadoop2.6.tgz，解压到/opt/spark-2.0.0-bin-hadoop2.7  
                    下载地址：http://spark.apache.org/downloads.html,  
  
             配置环境变量,编辑/etc/profile,执行以下命令       
python@ubuntu :~$  sudo nano /etc/profile  
                     在文件最增加：  
          \#Seeting JDK JDK环境变量  
            export JAVA\_HOME=/opt/jdk1.8.0\_45  
            export JRE\_HOME=$\{JAVA\_HOME\}/jre  
            export CLASSPATH=.:$\{JAVA\_HOME\}/lib:$\{JRE\_HOME\}/lib  
            export PATH=$\{JAVA\_HOME\}/bin:$\{JRE\_HOME\}/bin:$PATH           
         \#Seeting Scala Scala环境变量  
  
            export SCALA\_HOME=/opt/scala-2.11.8  
            export PATH=$\{SCALA\_HOME\}/bin:$PATH  
           export SPARK\_HOME=/opt/spark-2.0.0-bin-hadoop2.7  
         export PYTHONPATH=/opt/spark-2.0.0-bin-hadoop2.7/python  
  
     重启电脑，使/etc/profile永久生效，临时生效，打开命令窗口，执行 source /etc/profile  在当前窗口生效  
cd /opt/spark-2.0.0-bin-hadoop2.7  
./bin/spark-shell  
http://127.0.0.1:4040/executors/  -----------一定要开  
  
Scala和Python两种语言。Java不支持交互式的Shell，因此这一功能暂未在Java语言中实现。  
  
可以用spark-shell.cmd和pyspark.cmd命令分别运行Scala版本和Python版本的Spark Shell。  
Spark网页控制台  
  
不论Spark运行在哪一种模式下，都可以通过访问Spark网页控制台查看Spark的作业结果和其他的统计数据，控制台的URL地址如下：  
  
http://localhost:4040  
  
Spark控制台如下图3所示，包括Stages，Storage，Environment和Executors四个标签页  
./bin/pyspark  
,打开Scala到Spark的连接窗口     
  
  
nano customers.txt  
100, John Smith, Austin, TX, 78727  
200, Joe Johnson, Dallas, TX, 75201  
300, Bob Jones, Houston, TX, 77028  
400, Andy Davis, San Antonio, TX, 78227  
500, James Williams, Austin, TX, 78727  
  
下述代码片段展示了可以在Spark Shell终端执行的Spark SQL命令。  
  
// 首先用已有的Spark Context对象创建SQLContext对象  
val sqlContext = new org.apache.spark.sql.SQLContext(sc)  
  
// 导入语句，可以隐式地将RDD转化成DataFrame  
import sqlContext.implicits.\_  
  
// 创建一个表示客户的自定义类  
case class Customer(customer\_id: Int, name: String, city: String, state: String, zip\_code: String)  
  
// 用数据集文本文件创建一个Customer对象的DataFrame  
val dfCustomers = sc.textFile("data/customers.txt").map(\_.split(",")).map(p => Customer(p(0).trim.toInt, p(1), p(2), p(3), p(4))).toDF()  
  
// 将DataFrame注册为一个表  
dfCustomers.registerTempTable("customers")  
  
// 显示DataFrame的内容===god  
dfCustomers.show()  
  
// 打印DF模式  
dfCustomers.printSchema()  
  
// 选择客户名称列  
dfCustomers.select("name").show()  
  
// 选择客户名称和城市列  
dfCustomers.select("name", "city").show()  
  
// 根据id选择客户  
dfCustomers.filter(dfCustomers("customer\_id").equalTo(500)).show()  
  
// 根据邮政编码统计客户数量  
dfCustomers.groupBy("zip\_code").count().show()  
  
在上一示例中，模式是通过反射而得来的。我们也可以通过编程的方式指定数据集的模式。这种方法在由于数据的结构以字符串的形式编码而无法提前定义定制类的情况下非常实用。  
  
如下代码示例展示了如何使用新的数据类型类StructType，StringType和StructField指定模式。  
  
//  
// 用编程的方式指定模式  
//  
  
// 用已有的Spark Context对象创建SQLContext对象  
val sqlContext = new org.apache.spark.sql.SQLContext(sc)  
  
// 创建RDD对象  
val rddCustomers = sc.textFile("user/customers.txt")  
  
// 用字符串编码模式  
val schemaString = "customer\_id name city state zip\_code"  
  
// 导入Spark SQL数据类型和Row  
import org.apache.spark.sql.\_  
  
import org.apache.spark.sql.types.\_;  
  
// 用模式字符串生成模式对象  
val schema = StructType(schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))  
  
// 将RDD（rddCustomers）记录转化成Row。  
val rowRDD = rddCustomers.map(\_.split(",")).map(p => Row(p(0).trim,p(1),p(2),p(3),p(4)))  
  
// 将模式应用于RDD对象。  
val dfCustomers = sqlContext.createDataFrame(rowRDD, schema)  
  
// 将DataFrame注册为表  
dfCustomers.registerTempTable("customers")  
  
// 用sqlContext对象提供的sql方法执行SQL语句。  
val custNames = sqlContext.sql("SELECT name FROM customers")  
  
// SQL查询的返回结果为DataFrame对象，支持所有通用的RDD操作。  
// 可以按照顺序访问结果行的各个列。  
custNames.map(t => "Name: " + t(0)).collect().foreach(println)  
  
// 用sqlContext对象提供的sql方法执行SQL语句。  
val customersByCity = sqlContext.sql("SELECT name,zip\_code FROM customers ORDER BY zip\_code")  
  
// SQL查询的返回结果为DataFrame对象，支持所有通用的RDD操作。  
// 可以按照顺序访问结果行的各个列。  
customersByCity.map(t => t(0) + "," + t(1)).collect().foreach(println)