zoukankan      html  css  js  c++  java
  • Spark笔记

    Spark是一个开源的类似于Hadoop MapReduce的通用的并行计算框架,Spark基于map reduce算法实现的分布式计算,拥有Hadoop MapReduce所具有的优点;但不同于MapReduce的是Spark中的Job中间输出和结果可以保存在内存中,从而不再需要读写HDFS,因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的map reduce的算法。

    SparkMapReduce的替代方案,而且兼容HDFSHive,可融入Hadoop的生态系统,以弥补MapReduce的不足。

    特点 快 易用 通用 兼容性

    Spark Core实现了 Spark 的基本功能,包含任务调度、内存管理、错误恢复、与存储系统 交互等模块。Spark Core 中还包含了对弹性分布式数据集(resilient distributed dataset,简称RDD)API 定义。

    Spark SQL Spark 用来操作结构化数据的程序包。通过 Spark SQL,我们可以使用 SQL 或者 Apache Hive 版本的 SQL 方言(HQL)来查询数据。Spark SQL 支持多种数据源,比 如 Hive 表、Parquet 以及 JSON 等。

    Spark Streaming Spark 提供的对实时数据进行流式计算的组件。提供了用来操作数据流的 API,并且与 Spark Core 中的 RDD API 高度对应。

    Spark MLlib提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等,还提供了模型评估、数据 导入等额外的支持功能。

    集群管理器:Spark 设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计 算。为了实现这样的要求,同时获得最大灵活性,Spark 支持在各种集群管理器(cluster manager)上运行,包括 Hadoop YARNApache Mesos,以及 Spark 自带的一个简易调度 器,叫作独立调度器。

    spark运行模式:

    本地模式用于开发测试

    standalone:构建一个由master和salver构成的spark集群 spark在集群中运行

    spark on yarn:直连yarn,不需要额外构建spark集群

    spark on mesos:直连mesos,不需要额外构建spark集群

    spark任务提交:

    bin/spark-submit 
      --class <main-class>
      --master <master-url> 
      --deploy-mode <deploy-mode> 
      --conf <key>=<value> 
      ... # other options
      <application-jar> 
      [application-arguments]
    

     常用的参数

    1) --class: 你的应用的启动类 (org.apache.spark.examples.SparkPi)

    2) --master: 集群的master URL (spark://node01:7077)

    3) --deploy-mode: 是否发布你的驱动到worker节点(cluster) 或者作为一个本地客户端 (client) (default: client)*

    4) --conf: 任意的Spark配置属性, 格式key=value. 如果值包含空格,可以加引号“key=value”.

    5) application-jar: 打包好的应用jar,包含依赖. 这个URL在集群中全局可见。 比如hdfs:// 共享存储系统, 如果是 file:// path, 那么所有的节点的path都包含同样的jar.

    6) application-arguments: 传给main()方法的参数

    Master URL 可以是以下格式:

    查看Spark-submit全部参数:

     

    local

    本地以一个worker线程运行(例如非并行的情况).

    local[K]

    本地以K worker 线程 (理想情况下, K设置为你机器的CPU核数).

    local[*]

    本地以本机同样核数的线程运行.

    spark://HOST:PORT

    连接到指定的Spark standalone cluster master. 端口是你的master集群配置的端口,缺省值为7077.

    mesos://HOST:PORT

    连接到指定的Mesos 集群. Port是你配置的mesos端口, 缺省是5050. 或者如果Mesos使用ZOoKeeper,格式为 mesos://zk://....

    yarn-client

    client模式连接到YARN cluster. 集群的位置基于HADOOP_CONF_DIR 变量找到.

    yarn-cluster

    cluster模式连接到YARN cluster. 集群的位置基于HADOOP_CONF_DIR 变量找到.

     

    spark启动 spark-shell --master local[N]

     

     

    RDD(Resilient Distributed Dataset)叫做弹性分布式数据集,是Spark中最基本的数据抽象,它代表一个不可变、可分区、里面的元素可并行计算的集合。RDD具有数据流模型的特点:自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显式地将数据缓存在内存中,后续的查询能够重用这些数据,这极大地提升了查询速度。

     

    Dataset:一个数据集合,用于存放数据的。

     

    Distributed:RDD中的数据是分布式存储的,可用于分布式计算。

     

    Resilient:RDD中的数据可以存储在内存中或者磁盘中。

    一组分片(Partition),即数据集的基本组成单位。

    一个计算每个分区的函数。

    RDD之间的依赖关系。

    一个Partitioner,即RDD的分片函数。

    一个列表,存储存取每个Partition的优先位置(preferred location)。

    RDD特点 分区 只读 依赖 缓存 checkpoint

    创建RDD的三种方式 已有的集合 外部存储文件 已有的RDD经过算子转换

    RDD的算子分类

    Transformation(转换):根据数据集创建一个新的数据集,计算后返回一个新RDD;例如:一个rdd进行map操作后生了一个新的rdd

    Action(动作):对rdd结果计算后返回一个数值value给驱动程序;

    例如:collect算子将数据集的所有元素收集完成返回给驱动程序

    RDD中的所有转换都是延迟加载的,也就是说,它们并不会直接计算结果。相反的,它们只是记住这些应用到基础数据集(例如一个文件)上的转换动作。只有当发生一个要求返回结果给Driver的动作时,这些转换才会真正运行。这种设计让Spark更加有效率地运行。

     

     

    转换

    含义

    map(func)

    返回一个新的RDD,该RDD由每一个输入元素经过func函数转换后组成

    filter(func)

    返回一个新的RDD,该RDD由经过func函数计算后返回值为true的输入元素组成

    flatMap(func)

    类似于map,但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列,而不是单一元素)

    mapPartitions(func)

    类似于map,但独立地在RDD的每一个分片上运行,因此在类型为T的RDD上运行时,func的函数类型必须是Iterator[T] => Iterator[U]

    mapPartitionsWithIndex(func)

    类似于mapPartitions,但func带有一个整数参数表示分片的索引值,因此在类型为T的RDD上运行时,func的函数类型必须是

    (Int, Interator[T]) => Iterator[U]

    union(otherDataset)

    对源RDD和参数RDD求并集后返回一个新的RDD

    intersection(otherDataset)

    对源RDD和参数RDD求交集后返回一个新的RDD

    distinct([numTasks]))

    对源RDD进行去重后返回一个新的RDD

    groupByKey([numTasks])

    在一个(K,V)的RDD上调用,返回一个(K, Iterator[V])的RDD

    reduceByKey(func, [numTasks])

    在一个(K,V)的RDD上调用,返回一个(K,V)的RDD,使用指定的reduce函数,将相同key的值聚合到一起,与groupByKey类似,reduce任务的个数可以通过第二个可选的参数来设置

    sortByKey([ascending], [numTasks])

    在一个(K,V)的RDD上调用,K必须实现Ordered接口,返回一个按照key进行排序的(K,V)的RDD

    sortBy(func,[ascending], [numTasks])

    与sortByKey类似,但是更灵活

    join(otherDataset, [numTasks])

    在类型为(K,V)和(K,W)的RDD上调用,返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

    cogroup(otherDataset, [numTasks])

    在类型为(K,V)和(K,W)的RDD上调用,返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

    coalesce(numPartitions)

    减少 RDD 的分区数到指定值。

    repartition(numPartitions)

    重新给 RDD 分区

    repartitionAndSortWithinPartitions(partitioner)

     

    重新给 RDD 分区,并且每个分区内以记录的 key 排序

     

    动作

    含义

    reduce(func)

    reduce将RDD中元素前两个传给输入函数,产生一个新的return值,新产生的return值与RDD中下一个元素(第三个元素)组成两个元素,再被传给输入函数,直到最后只有一个值为止。

    collect()

    在驱动程序中,以数组的形式返回数据集的所有元素

    count()

    返回RDD的元素个数

    first()

    返回RDD的第一个元素(类似于take(1))

    take(n)

    返回一个由数据集的前n个元素组成的数组

    takeOrdered(n, [ordering])

    返回自然顺序或者自定义顺序的前 n 个元素

    saveAsTextFile(path)

    将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本

    saveAsSequenceFile(path) 

    将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统。

    saveAsObjectFile(path) 

    将数据集的元素,以 Java 序列化的方式保存到指定的目录下

    countByKey()

    针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。

    foreach(func)

    在数据集的每一个元素上,运行函数func

    foreachPartition(func)

    在数据集的每一个分区上,运行函数func

  • 相关阅读:
    C代码优化的一些方法
    读书笔记之:C++程序设计原理与实践(其他)[+++]
    读书笔记之:C++ Primer Plus(第5版)
    使用SoapFormatter 类序列化
    C# 多线程学习系列
    winform中读取嵌入的xml文件
    使用Web Service上(服务器端访问)
    jquery ajax调用webservice
    ashx文件和HttpHandler
    以Window服务作为wcf的寄主
  • 原文地址:https://www.cnblogs.com/lishisan/p/10665738.html
Copyright © 2011-2022 走看看