第一章 HDFS架构

1 Master（NameNode/NN）    带N个Slaves（DataNode/DN）

1个文件会被拆成多个Block

NN：

DN：

典型策略：

replication factor：副本系数、副本因子

All block in a file except the last block are the same size.

linux命令

   wget  http://archive.cloudera.com/cdh5/cdh/5/hadoop-2.6.0-cdh5.7.0.tar.gz

2) 安装jdk

    修改机器名：vi /etc/sysconfig/network

    设置ip和hostname的映射关系：/etc/hosts

    在CentOS7中要修改主机名称（hostname）只能修改/etc/hostname文件内容来进行，修改/etc/sysconfig/network文件根本不起作用。

    ssh免密码登录：ssh-keygen -t rsa

4)hadoop安装配置

    解压：tar -zxvf hadoop.... -C /app

   Hadoop配置文件修改：/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop

   vi core-site.xml 

   <property>

5)格式化HDFS

6)启动HDFS

        注意：datanode当时没有启动成功，/etc/hosts中127.0.0.1没有和localhost对应

        浏览器验证：http://172.20.10.2:50070/

7）停止HDFS

    HDFS常用的shell命令：

NodeManager职责：整个集群中有N个，负责单个节点的资源管理和使用以及task的运行情况

yarn-site.xml

启动yarn：sbin/start-yarn.sh

验证启动是否成功：

    浏览器：http://172.20.10.3:8088

1）通常用于离线数据处理（采用MapReduce）

Hive底层执行引擎有：MapReduce\Tez\Spark

    Hive on MapReduce

压缩：GZIP、LZO、Snappy、BZIP2..

为什么要使用Hive？

Hive环境搭建：

2）解压

3）配置

环境变量设置：vi ~/.bash_profile 

export HIVE_HOME=/app/hive-1.1.0-cdh5.7.0

先进入/app/hive../cnf下复制：cp hive-env.sh.template hive-env.sh

进入mysql：/usr/bin/mysql -u root -p   密码：root

进入/app/hive../cnf新建文件hive-site.xml并写入：

<property>

<property>

<property>

4）拷贝mysql驱动到lib： cp mysql-connector-java-5.1.47-bin.jar /app/hive-1.1.0-cdh5.7.0/lib

5)启动hive：bin目录 ./hive   (前提：hadoop:./start-all.sh)

创建表：

加载数据到hive表：

load data local inpath '/data/hello1.txt' into table hive_wordcount1;

select word,count(1) from hive_wordcount lateral view explode(split(context,' ')) wc as word group by word;

select word,count(1) from hive_wordcount1 lateral view explode(split(context,' ')) wc as word group by word;

hive ql提交执行以后会生成mr作业，并在yarn上运行。

产生背景、概述及特点、发展历史、Spark Survey、Spark对比Hadoop、Spark和Hadoop的协作性、Spark开发语言、Spark运行模式

官网：spark.apache.org

Spark特点：

Speed(基于内存（避免数据落地）、DAG执行引擎、基于线程模型对比进程)、Ease of Use（语言丰富、API多）、Generality、Runs Everywhere

MapReduce的局限性：

Spark生态圈：BDAS（Berkeley Data Analytics Stack）

协作性：

Hadoop优势：

====》Spark

实战环境搭建：

Spark源码编译：目的是和指定版本的hadoop兼容使用。

前置要求：

<properties>

spark两者编译方式： 

方式一：mvn编译命令：

    ./build/mvn -Pyarn -Phadoop-2.6 -Phive -Phive-thriftserver -Dhadoop.version=2.6.0-cdh5.7.0 -DskipTests clean package

./dev/make-distribution.sh --name 2.6.0-cdh5.7.0 --tgz -Pyarn -Phadoop-2.6 -Phive -Phive-thriftserver -Dhadoop.version=2.6.0-cdh5.7.0

编译完成后：

    spark-2.1.0-bin-2.6.0-cdh5.7.0.tgz

坑一：

    [ERROR] Failed to execute goal on project spark-launcher_2.11: Could not resolve dependencies for project org.apache.spark:spark-launcher_2.11:jar:2.1.0: Failure to find org.apache.hadoop:hadoop-client:jar:2.6.0-cdh5.7.0 in http://maven.aliyun.com/nexus/content/groups/public was cached in the local repository, resolution will not be reattempted until the update interval of nexus-aliyun has elapsed or updates are forced -> [Help 1]

    需要pom.xml添加：

坑二：cd dev/===>vi make-distribution.sh(应该是这个位置吧？？)

    export MAVEN_OPTS=“-Xmx2g -XX:ReservedCodeCacheSize=512m -XX:MaxPermSize=512M”

    注意点：如果提示内存不够，可能vm需要调整内存至少2-4G

坑三：

    如果编译的是scala版本是2.10

坑四：

    两种方案解决：

    1)    去仓库把xxx.lastUpdated文件全部删除，重新执行maven命令

坑五：

    将编译好的spark压缩包解压到某个位置==〉配置spark环境变量～./bash_profile

    第一种：Local模式搭建：bin目录下执行==》spark-shell --master local[2]

    第二种：standalone模式搭建：

     cp spark-env.sh.template spark-env.sh 

     vi spark-env.sh

     添加：

        集群的搭建：

            conf目录下：slaves文件配置：

            hadoop1:master

            slaves:

            =====>start-all.sh   会在hadoop1机器上启动master进程，在slaves文件配置的所有hostname的机器上启动worker进程

    bin目录下：spark-shell --master spark://hadoop001:7077

    spark WordCount统计：

        val file=spark.sparkContext.textFile("file:///data/hello.txt") 

        val wordCounts=file.flatMap(line=>line.split(",")).map((word=>(word,1))).reduceByKey(_+_)

        wordCounts.collect

    Spark SQL前世今生

        为什么需要SQL？===》1）事实上的标准。 2）易学易用。 3）受众面广。 

        Spark:hive on spark===>shark（初期）

            shark推出：欢迎。基于spark、基于内存的列式存储、与hive能兼容

            缺点：hive ql的解析、逻辑执行计划生成、执行计划的优化是依赖于hive的

                仅仅只是把物理执行计划从mr作业替换成spark作业。

                依赖过大导致添加新功能不方便；mr基于进程级别处理，spark基于线程==》线程安全问题

        Shark终止以后，产生两个分支：

            1）Hive on Spark（Help existing Hive users migrate to spark）

                hive社区，源码在Hive中

            2）Spark SQL（A new SQL engine designed from ground-up for Spark）

                Spark社区，源码在Spark中

                    优点：支持多种数据源，多种优化技术，扩展性好很多。

        1）Hive

            sql==》mapreduce

            metastore：元数据

            sql：database、table、view

        2）impala

            cloudera：cdh(建议大家在生产上使用的hadoop系列版本)

            sql:自己的守护进程执行的，非mr

            metastore

        3）presto

            facebook   京东   sql

        4）drill

                sql  支持访问：hdfs、rdbms、json、hbase、mangodb、s3、hive

        5）Spark SQL

                sql  支持访问：hdfs、rdbms、json、hbase、mangodb、s3、hive

                基于dataframe/dataset api 编程

                metastore

            1）Part of the core distributio since Spark1.0(April 2014)

            2)Runs SQL/Hive Ql queries including UDFs UDAFs and SerDes

            3)Connect existing BI tools to Spark througth JDBC

            4)Binding in Python,Scala,Java,R

        Spark SQL is Apache Spark's module for working with structured data.

        有见到SQL字样吗？

        Spark SQL概述小结：

            1）Spark SQL的应用不局限于SQL

            2）访问hive、json、parquet等文件的数据

            3）SQL只是Spark SQL的一个功能

            4）Spark SQL提供了SQL的api、DataFrame和Dataset的API

        Write less code

        Read less data

        Let the optimizer do the the hard work

    Spark SQL架构

        见架构图

第六章 从Hive平滑过渡到Spark SQl

    SQLContext/HiveContext/SparkSession的使用==>前两个已经过时

    spark-shell/spark-sql的使用

    thriftserver/beeline的使用

    jdbc方式编程访问

     创建maven项目===》构建ImoocSparkSQLProject项目

    package com.imooc.spark

    object SQLContextApp {

        val path=args(0);

        //1）创建相应的context

        //在测试和生产环境中，AppName和Master建议在脚本中指定

        val sc=new SparkContext(sparkConf)

        //2）相关的处理

    注意：path参数的传入，点击项目Edit Configurations

    打包执行：

        打好的jar包在target文件夹下，上传到服务器上：sudo scp  1.jar root@172.20.10.3:~/lib

    执行以下代码执行：

      spark-submit \

          vi sqlcontext.sh

          注意赋给用户该文件执行的权限：chmod u+x sqlcontext.sh

          执行./sqlcontext.sh

     spark-submit \

      区别：需要jdbc驱动，找到元数据，不需要传入参数。

SparkSession:

          def main(args: Array[String]): Unit = {

        val spark=SparkSession.builder().appName("SparkSessionApp").master("local[2]").getOrCreate()

        spark.read.json("file:///imooc/people.json").show()

        spark.stop()

spark-shell/spark-sql的使用：

        前提条件：1）在spark/conf中添加hive/conf中的hive-setting.xml文件，以便连接数据库；

        =====》对比spark查询表和hive查询表速度上的区别。

        spark-sql --master local[2] --jars /app/mysql-connector-java-5.1.30-bin.jar

        explain extended select a.key*(2+3),b.value from t a join t b on a.key=b.key and key>3; 

        == Parsed Logical Plan ==

        == Analyzed Logical Plan ==

        == Optimized Logical Plan ==

        == Physical Plan ==

thriftserver/beeline的使用

        =====》实际开发中最常用的方式

1）启动thriftserver：默认端口是10000，可以修改

    spark/sbin目录下：

        ./start-thriftserver.sh --master local[2] --jars /app/mysql-connector-java-5.1.30-bin.jar

    使用：jps -m 查看是否启动成功？

    beeline -u jdbc:hive2://localhost:10000 -n hadoop

    修改thriftserver启动占用的默认端口号：

    ./start-thrifserver.sh \

    --master local[2] \

    --jars /app/mysql-connector....

    --hiveconf hive.server2.thrift.port=1400

    beeline -u jdbc:hive2://localhost:14000 -n hadoop

    thriftserver和普通的spark-shell/spark-sql有什么不同？

    1）spark-shell、spark-sql都是一个spark application；

    2）thriftserver，不管你启动多少个客户端（beeline/code），永远都是一个spark application

        解决了一个数据共享的问题，多个客户端可以共享数据；

        注意：一定先要启动thriftserver

        import java.sql.DriverManager

        object SparkSQLThriftserverApp {

            Class.forName("org.apache.hive.jdbc.HiveDriver")

             val con=DriverManager.getConnection("jdbc:hive2://172.20.10.3:10000","","")

    DataFrame产生背景

        Spark RDD API vs MapReduce API

        R/Pandas（最早产生DataFrame，可以无缝连接）

    DataFram概述

        DataFrame它不是spark sql提出的，而是最早在R、Pandas语言就已经有了的。

        A Dataset is a distributed collection of data:分布式的数据集

        A DataFrame is a Dataset organized into named columns

        以列（列名、列的类型、列值）的形式构成的分布式数据集，按照列赋予不同的名称。

    DataFrame对比RDD

        图：

        RDD：

        DataFrame：

        ====〉执行性能不同。

    DataFrame基本API常用操作

        object DataFrameApp {

            val spark=SparkSession.builder().appName("DataFrameApp").master("local[2]").getOrCreate()

            //输出DataFrame对应的schema信息

            //输出数据集前20条记录

            //查询某列数据：select name from table

            //查询某几列数据，并对列进行计算：select name,(age+10) as age2 from table

            //根据某一列进行分组，再进行聚合操作

            //根据年龄进行过滤操作

            spark.stop() 

        package com.imooc.spark

        import org.apache.spark.sql.SparkSession

        object DataFrameCase {

           val spark=SparkSession.builder().appName("DataFrameCase").master("local[2]").getOrCreate()

            val rdd=spark.sparkContext.textFile("file:///imooc/student.txt")

          import spark.implicits._

        //    studentDF.show(20,false)

            studentDF.filter("substring(name,0,1)='A'").show()

        //    studentDF.select(studentDF.col("name").as("student_name")).show()

          }

        package com.imooc.spark

            val spark=SparkSession.builder().appName("DataFramRDD").master("local[2]").getOrCreate()

            program(spark)

            spark.close()

            //RDD==>DataFrame

            // 注意需要导入隐式转换

            infosDF.show()

            //将DataFrame转化成sql可以处理的表

            spark.sql("select * from info where age>30").show()

          case class Info(id:Int, name:String, age:Int)

        DataFrame和RDD两种互操作的区别：

        2）编程：Row。  如果第一宗情况不能满足你的要求

        3）优先选择第一种互操作

        object DatasetApp {

            //如何解析csv文件？

第八章 External Data Source API

    产生背景

        Every Spark application starts with loading data and ends with saving data.

        用户：方便快速从不同的数据源（json、parquet、rdbms），经过混合处理（json join parquet），

            再将处理结果以特定的格式（json/parquet）写回到指定的系统（HDFS、S3）上去。

        ==》Spark SQL 1.2===〉外部数据源API

    Loading and saving data is not easy:

        Parse raw data:text/json/parquet

        Convert data forma transformation

        Datasets stored in various Formats/System.

    目标

        外部数据源的目的

            1）开发人员：是否需要把代码合并到spark中？？不需要。

            2）用户

                写：people.write.format("parguet").save("path")

    操作Parquet文件数据

        spark-shell处理：

            object ParquetApp {

            df.select("name","favorite_color").show()

          }

        spark-sql处理：

            create temporary view parquetTable using org.apache.spark.sql.parquet options(path "/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/users.parquet");

            select * from parquetTable;

        spark.sql("select deptno,count(1) as mount from emp where group by deptno").filter("deptno is not null").write.saveAsTable("hive_table_1")

        如果不加as mount：

        spark.sqlContext.setConf("spark.sql.shuffle.partitions",10)

    操作Mysql表数据

        第一种方式：

        val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/")

        第二种方式：

        import java.util.Properties

        val jdbcDF2 = spark.read.jdbc("jdbc:mysql://localhost:3306/", "mysql.tables_priv", connectionProperties)

    1）用户行为日志概述

        用户行为日志：用户每次访问网站时所有的行为数据（访问、浏览、搜索、点击...）

        又称：用户行为轨迹、流量日志

        用户行为日志生成渠道：Nginx和Ajax

        用户行为日志内容：

            一次访问记录：

            访问者所有的客户端（手机/电脑，chrome、firefox..）、访问模块id、跳转/连接地址...

    2）离线数据处理架构

        1、数据采集

        2、数据清洗

        3、数据处理

        4、处理结果入库

        5、数据的可视化

        一般的日志处理方式，我们是需要进行分区的，

    3）项目需求

        需求一：统计imooc主战最受欢迎的课程/手记的Top N访问次数

        需求二：按地市统计imooc主站最受欢迎的Top N课程

        需求三：按流量来统计imooc主站最受欢迎的Top N课程

    4）功能实现

        输入：访问时间、访问URL、访问过程耗费流量、访问IP地址

        输出：URL、cmsType（video/article）、cmsId（编号）、流量、ip、城市信息、访问时间、分区（天）

        坑：

            case:Exception => Row(0)的问题 换成Row("","",0L,0L,"","","","")

        使用github上已有的开源项目

        2)编译已经下载的项目：mvn clean package -DskipTests

        3)安装jar包到自己的maven仓库

            mvn install:install-file -Dfile=/imooc/BigDataImooc/ipdatabase/target/ipdatabase-1.0-SNAPSHOT.jar -DgroupId=com.ggstar -DartifactId=ipdatabase -Dversion=1.0 -Dpackaging=jar    

            坑：maven报错 the goal you specified requires a project to execute but there is no POM in this direct...

            这个错误的原因是当前目录下没有pom.xml 文件,Maven执行必须要pom.xml文件，我就跑到pom.xml文件目录下 mvn install.问题解决

        4)idea项目pom.xml添加依赖

            <dependency>

        5)清洗后数据保存

            df.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite).partitionBy("day").save("/imooc/clean/")

            按字段”day“分区==》分区后每个区执行文件个数coalesce()==>mode模式可以选择覆盖

        6）写入数据库Mysql

            create table day_video_access_topn_stat(

            create table day_video_city_access_topn_stat(

            //加入主键出问题。

        数据可视化：一幅图片最伟大的价值莫过于它能够使的我们实际看到的比我们期望看到的内容更加丰富

        常用的数据可视化框架：echarts、highcharts、D3.js。

            不需要开发的框架：HUE、Zeppelin

        本次课程使用echarts和Zeppelin作数据可视化。

        spark支持可插拔的集群管理模式

            在spark中，支持4种运行模式：

            1、Local:开发时使用

            2、Standalone：

            3、Yarn：建议大家在生产上使用该模式，统一使用Yarn进行整个集群作业（MR、spark）的资源调度

            4、Mesos:国内基本不用

            不管使用什么模式，spark应用程序的代码都是一样的，只需要在提交的时候通过--master参数来指定我们的运行模式即可。

            spark on yarn之client模式：

                Driver运行在Client端（提交spark作业的机器）

            spark on yarn之cluster模式：

                Driver运行在ApplicatinMaster中

                client模式：

                //注意格式要求

                ./bin/spark-submit --class org.apache.spark.examples.SparkPi  --master yarn --executor-memory 1G  --num-executors 1  /app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/jars/spark-examples_2.11-2.1.0.jar  4

                 此处的yarn就是client模式,如果是cluster模式的话，就写 yarn-cluster.cd

                 报错：

                 Exception in thread "main" java.lang.Exception: When running with master 'yarn' either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.

                 如果想运行在YARN之上，那么就必须要设置HADOOP_CONF_DIR或者YARN_CONF_DIR

                 1、export HADOOP_CONF_DIR=/app/hadoop-2.6.0-cdh5.7.0/etc/hadoop

                ./bin/spark-submit \

                  ./bin/spark-submit --class org.apache.spark.examples.SparkPi  --master yarn-cluster --executor-memory 1G  --num-executors 1  /app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/jars/spark-examples_2.11-2.1.0.jar  4

                  命令查看日志：==》还是有报错的

        <plugin>

        打包命令：mvn assembly:assembly

                ./bin/spark-submit \

        集群优化：1、存储格式的选择：www.infoq.com/cn.article/bigdata-store-choose/

                 2、压缩格式的选择：www.ibm.com/developerworks/cn/opensource/os-cn-hadoop-compression-analysis/

                     压缩速度和压缩文件的可分割性

                 默认格式是snappy，可选择合适压缩格式：

                 SparkSession.builder().config("spark.sql.parquet.compression.codec","gzip").getOrCreate()

        代码优化：1、选择高性能的算子

                     将数据写入数据库的统计操作：cityTopStatDF.foreachPartition(partOfRecord=>{}、list（每个分区）、 pstmt.addBatch()等

                 2、复用已有的代码/数据

                    一般将其设置为false：