zoukankan      html  css  js  c++  java
  • SparkConf和SparkContext

    任何Spark程序都是SparkContext开始的,SparkContext的初始化需要一个SparkConf对象,SparkConf包含了Spark集群配置的各种参数。
    初始化后,就可以使用SparkContext对象所包含的各种方法来创建和操作RDD和共享变量。
    Scala:
    val conf = new SparkConf().setMaster("master").setAppName("appName")
    val sc = new SparkContext(conf)
    或者
    val sc = new SparkContext("master","appName")
     

    通过创建SparkConf对象来配置应用,然后基于这个SparkConf创建一个SparkContext对象。驱动器程序通过SparkContext对象来访问Spark。

    这个对象代表对计算集群的一个连接。一旦有了SparkContext, 就可以用它来创建RDD。

    Java:

    SparkConf conf = new SparkConf().setMaster("local").setAppName("My App");
    JavaSparkContext sc = new JavaSparkContext(conf);

    只需传递两个参数:
    集群URL:告诉Spark如何连接到集群上。例子中使用local可以让Spark运行在单机单线程上而无需连接到集群。
    应用名:例子中使用My App。当连接到一个集群时,这个值可以帮助我们在集群管理器的用户界面中找到应用。

    在Spark2.0之前, SparkContext 是所有 Spark 功能的结构, 驱动器(driver) 通过SparkContext 连接到集群 (通过resource manager), 因为在2.0之前, RDD就是Spark的基础。

    如果需要建立SparkContext,则需要SparkConf,通过Conf来配置SparkContext的内容。

    在Spark2.0之后,Spark Session也是Spark 的一个入口, 为了引入dataframe和dataset的API, 同时保留了原来SparkContext的functionality,

    如果想要使用 HIVE,SQL,Streaming的API, 就需要Spark Session作为入口。

    SparkSession spark = SparkSession.builder().appName("demo_spark").enableHiveSupport().getOrCreate();
    

    1)SparkSession.builder() 创建此方法用于构造SparkSession。

    2)master(“local”) 设置要连接的master URL,例如:

    “local”在本地运行
    “local[4]”以4核在本地运行
    “spark://master:7077”在spark独立集群上运行

    3)appName( ) 设置将在spark Web UI中显示的应用程序的名称。如果未设置应用程序名称,则将使用随机生成的名称。

    4)Config 设置使用此方法设置的配置选项会自动传递到'SparkConf'和'SparkSession'自己的配置,它的参数由键值对组成。

    5)enableHiveSupport启用Hive支持,类似于HiveContext创建了sparkSession,我们可以用它来读取数据。

    6)getOrCreate()方法表示有就拿过来,没有就创建,类似于单例模式。

    使用SparkSession读取数据SparkSession是读取数据的入口点,类似于旧的SQLContext.read。以下代码使用SparkSession从CSV读取数据:

     val df = spark.read.format("com.databricks.spark.csv")                 
    .schema(customSchema)                   
    .load("data.csv")

      

    从Spark 2.0.0开始,最好使用SparkSession,因为它提供了对sparkContext所具有的所有spark功能的访问。 此外,它还提供了用于处理DataFrame和DataSet的API

    运行SQL查询SparkSession可用于对数据执行SQL查询,将结果作为Data-Frame(即数据集[ROW])返回。

    众所周知,在以前的版本中,sparkcontext 是spark的入口点,因为RDD是主要的API,它是使用上下文API创建和操作的。 对于每个其他API,我们需要使用不同的context。

    对于流式传输,我们需要streamingContext。 对于SQL sqlContext和hive hiveContext.,因为dataSet和DataFrame API正在成为新的独立API,我们需要为它们构建入口点。 因此在spark 2.0中,我们为DataSet和DataFrame API创建了一个新的入口点构建,称为Spark-Session。

     它是SQLContext,HiveContext和未来的streamingContext的组合。 在这些context中可用的所有API都可以在SparkSession上获得,SparkSession也有实际计算的spark context 。

     SparkSession: SparkSession实质上是SQLContext和HiveContext的组合(未来可能还会加上StreamingContext),

                                   所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext,所以计算实际上是由sparkContext完成的。

    Scala:

    val sparkSession = SparkSession.builder
                        .master("master")
                        .appName("appName")
                        .getOrCreate()
    或者
    SparkSession.builder.config(conf=SparkConf())
    

      

    I、 SparkSubmit 的 shell脚本

    /data/spark/spark-2.2.0-bin-hadoop2.7/bin/spark-submit

    --master spark://elcndc2sc39t:7077

    --class com.enc.analysis.core.AlgorithmExecute

    /data/upload/analysis/analysisFrame-1.1.0.jar $1 $2

    -----------------------------------------------------------------

    --master表示master路径,

    --class表示入口的类的全路径

    /data/upload/analysis/analysisFrame-1.1.0.jar 表示计算框架jar包的全路径

    $1,$2..是自定义的shell命令进行传参,传递的参数会在入口类的main方法的String[] args中

    II、 利用Spark读取jdbc

    Properties connectionProperties = new Properties();
    
    String url = "jdbc:mysql://" + "mysql服务器地址" + ":" + "mysql端口" + "/" + "数据库名?useUnicode=true&characterEncoding=utf-8";
    
    String driver = "com.mysql.jdbc.Driver";
    
    connectionProperties.setProperty("user", "用户名");// 设置用户名
    
    connectionProperties.setProperty("password", "密码");// 设置密码
    
    connectionProperties.setProperty("driver", driver);
    
    connectionProperties.setProperty("url",url);
    
    SparkSession spark = SparkSessionUtils.getLocalSession();
    
    Dataset<Row> dataset = spark.read().jdbc(connectionProperties.getProperty("url"),"表名",connectionProperties).persist();
    
    dataset.show();
    

      

  • 相关阅读:
    【spring data jpa】jpa中criteria拼接in查询
    【spring boot】spring boot中使用@RestController不起作用,不返回json,依旧去找访问接口的请求地址对应的页面
    【mysql】mysql查询 A表B表 1对多 统计A表对应B表中如果有对应,则返回true否则false作为A表查询结果返回
    【redis】spring boot中 使用redis hash 操作 --- 之 使用redis实现库存的并发有序操作
    【多线程】java多线程Completablefuture 详解【在spring cloud微服务之间调用,防止接口超时的应用】【未完成】
    【docker】docker network常用命令参数
    【mysql】二级索引----聚簇索引和非聚簇索引-----
    【mysql】mysql统计查询count的效率优化问题
    SpringUtils
    idea激活
  • 原文地址:https://www.cnblogs.com/Allen-rg/p/11364683.html
Copyright © 2011-2022 走看看