跟天齊老師學Spark（3）--搭建Spark環境

2019-11-11 04:28:57

字體：大中小

來源：轉載

供稿：網友

搭建spark環境（偽分布式，在虛擬機上用一臺機器來編寫一個spark程序）：Spark解壓就能用。首先spark環境，首先要按照jdk，但可以不用按照scala環境。下載的spark最好是和hadoop版本匹配的spark版本，比如：spark-1.6.2-bin-hadoop2.6.tgz;需要自己使用maven來編譯（后面會講）。使用bin目錄下的spark-shell，這是spark的一個客戶端?？梢詥右粋€spark的單機版。它啟動之后會創建一個SparkContext對象或實例。啟動后會進入scala的命令行。spark中的方法（或算子）是延遲加載的。他的算子有兩鐘，其中有一種就是延遲加載的，一旦觸發Action的時候，它才會加載數據；比如：sc.textFile("/root/Words.txt")//延遲加載它默認使用hdfs的接口來讀的。但是這個textFile方法會把key丟掉，只保留value，因為那個key沒有用。sc.textFile("/root/words.txt").collect//收集結果sc.textFile("/root/words.txt").flatMap(_.split)這個flatMap是spark的rdd上的一個方法，和原來scala原生的flatMap方法的實現是不一樣的，但是功能都是一樣的。sc.textFile("/root/words.txt").flatMap(_.split(" ")).map(_, 1)它會生成一個RDD，這是一個分布式的數據集合。sc.textFile("/root/words.txt").flatMap(_.split(" ")).map(_, 1).reduceByKey(_+_)這個reduceByKey現在一個分區聚合，它會把value累加。到現在，這些方法都是延遲加載的，現在還沒有觸發任務。sc.textFile("/root/words.txt").flatMap(_.split(" ")).map(_, 1).reduceByKey(_+_).collect//收集結果這個時候會觸發Action，開始計算，得到結果。sc.textFile("/root/words.txt").flatMap(_.split(" ")).map(_, 1).reduceByKey(_+_).sortBy(_._2, false).collect這些方法和scala中的那些原生的方法是不一樣的，因為spark的這些方法都是針對分布式數據集合的。sc.textFile("/root/words.txt").flatMap(_.split(" ")).map(_, 1).reduceByKey(_+_).sortBy(_._2, false).saveAsTextFile("/root/out333")可以將計算結果保存到本地目錄上。Spark的輸入默認用的hdfs的InputFormat，輸出默認用的也是hdfs的OutputFormat。------------------------------------將多有節點的時間做一個同步：date -s "016-08-15 16:00:00"部署Spark集群：一個Master，多個Worker。老大負責資源調度，小弟負責節點管理，真正負責干活的是Executor（執行器）。這是spark自帶的一種集群模式：standalone。啟動一個老大，小弟們和老大建立連接，向老大注冊。提交任務的是Driver(客戶端)：跟老大建立通信，Master進行資源分配，讓Worker啟動Executor，然后Executor會和Driver通信；現在我們部署3個節點：一個Master和兩個Worker。不建議使用root用戶，實際生產中可能會有權限問題。Apache提供了一些自動化部署工具，可以實現spark、hadoop、storm的自動部署。當然我們也可以寫一些自動部署的shell腳本。spark的目錄： sbin：啟動spark相關的命令，一般不加到classpath； bin：hadoop和yarn相關的操作命令，一般會加到classpath下。 1.安裝jdk；2.上傳spark安裝包；3.解壓spark安裝包；4.修改配置文件spark-env.sh.template 改成spark-env.shslaves.template改成slaves（1）設置java_HOME; export JAVA_HOME=（2）告訴它Master的地址和端口。 vi spark-env.sh export SPARK_MASTER_ip=hadoop01 export SPARK_MASTER_PORT=7077 slaves hadoop02 hadoop035.將spark的配置拷貝到其他機器上 scp -r /itcast/hadoop02:/ scp -r /itcast/hadoop03:/ 這樣spark就部署好了。啟動spark（提前配置ssh免密碼登錄）: 在Master機器上啟動即可：sbin/start-all.sh通過jps查看進程，master機器上的進程為Master，其他兩臺機器都是Worker進程。spark只有兩個進程，一種是Master進程，一種是Worker進程。spark的管理界面地址：http://hadoop01:8080---------------------------------------上面的部署存在單點故障，需要部署兩個Master。多個Master和多個Worker之間的協調就要依靠zookeeper。Master啟動之后連接zookeeper，zookeeper會通過協調讓其中一個Master成為活躍的。而Worker啟動后也連接zookeeper，間接的知道哪個是活著的Master。所以要部署高可用的spark集群，就需要安裝zookeeper集群。在spark-env.sh中保留JAVA_HOME的配置，但是不要在配置Master的地址和端口號了，只要配置zookeeper的位置即可：export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=hadoop01:2181,hadoop02:2181,hadoop03:2181 -Dspark.deploy.zookeeper.dir=/sp"分別在3臺機器上啟動zookeeper，查看zookeeper的運行狀態。在hadoop01上啟動Master，使用spark/sbin/start-all.sh，這樣就會啟動一個Master和所有的Worker。然后在hadoop02上啟動一個Master，使用spark/sbin/start-master.sh命令（這次就不用再使用start-all.sh命令了）啟動。在spark的管理界面可以看到，一個Master是活躍（alive）狀態，另一個Master是備用（standby）狀態。Worker只會向活著的Master匯報，不會向Standby狀態的Master匯報。所以在hadoop02這個Master的管理界面上沒有Worker的信息。-----------------------------------------------------------spark/bin/spark-shell這個命令會啟動一個單機版的客戶端它沒有連到我們的集群上，它會啟動一個sparksubmit程序，它就相當于一個spark的客戶端，它要跟我們的集群通信，它可以把任務提交到集群上。啟動集群版的spark-shell的方法：在啟動的時候指定一些參數，現在我們有兩個Master，所以兩個Master的地址和端口都要寫（多個Master之間用逗號分隔）。spark/bin/spark-shell --master spark://hadoop01:7077,spark://hadoop02:7077注意：如果在啟動spark-shell的時候，你沒有給它指定要分配多少內存和CPU，那么它默認會給每一個Worker分配1G的內存，分配所有的核數。啟動成功后，在Worker所在的機器上就會分別啟動一個Executor進程；在scala的命令行使用sc.PRarllelize(Array(12,3,5,3,6,3,5,7))可以將這些數分到多臺機器上，RDD--->彈性的可復原的分布式的數據集。退出shell界面后，sparksubmit進程就會消失，同時所有的Worker機器上的Executor進程也會消失。補充：啟動spark-shell時指定cpu和內存的命令(根據任務來個性化分配資源)spark/bin/spark-shell --master spark://hadoop01:7077,spark://hadoop02:7077 --executor-memory 512m --total-executor-cores 7

上一篇：文章標題

下一篇：numpy