3开发环境搭建和开发示例,Spark的那些事

2019-10-01 14:36 来源:未知

上文提到,使用kudu等列式存储将数据以update模式写入kudu.
下面说一下java操作kudu的相关demo。java操作kudu在git上有相关demo,而spark操作kudu并没有。cloudera官网的操作中只提到了scala版本。本文列举java操作kudu的全示例,仅供入门参考。(痛苦的是sparksql查询kudu的java实现,官方没有示例,google也不好用)

【原创】Helenykwang 于2018-01-13 18:10:18编写

spark 2.X与1.x的区别

spark sql 2.x以上版本和1.x版本有个很大的区别:spark1.x的sqlContext在spark2.0中被整合到sparkSession,故而利用spark-shell客户端操作会有些许不同,具体如下文所述。


1)pom依赖

不用maven,不用sbt,只需三个库

载入外部数据的load方法

在spark sql中有一个DataStreamReader封装了读取各种格式的外部数据的方法,其中,format(str)用于传数据格式,比如csv,json,parquet,jdbc等;load(path)用于传入数据的地址,其中可以传入本地数据路径也可以是hdfs上的路径,在官网给的demo中都是传的本地数据路径:比如:

val usersDF = spark.read.load("examples/src/main/resources/users.parquet")
  • load(path)的源码:注意:load不能l载入hive的数据,hive数据需要使用table方法来载入。

    def load(path: String): DataFrame = {
     option("path", path).load()
       }
    
       def load(): DataFrame = {
     if (source.toLowerCase(Locale.ROOT) == DDLUtils.HIVE_PROVIDER) {
       throw new AnalysisException("Hive data source can only be used with tables, you can not " +
         "read files of Hive data source directly.")
     }
    
     val dataSource =
       DataSource(
       sparkSession,
     userSpecifiedSchema = userSpecifiedSchema,
     className = source,
     options = extraOptions.toMap)
     Dataset.ofRows(sparkSession, StreamingRelation(dataSource))
       }
    
  • 【hdfs路径】写入写出hdfs上的路径,则需要加入hdfs的完全路径,如:

    studentDF.write.parquet("hdfs://h4:9000/test/spark/parquet")
    studentDF.write.json("hdfs://h4:9000/test/spark/json")


  <!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-client -->
        <dependency>
            <groupId>org.apache.kudu</groupId>
            <artifactId>kudu-client</artifactId>
            <version>1.5.0-cdh5.13.1</version>
            <scope>test</scope>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-client-tools -->
        <dependency>
            <groupId>org.apache.kudu</groupId>
            <artifactId>kudu-client-tools</artifactId>
            <version>1.5.0-cdh5.13.1</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-spark2 -->
        <dependency>
            <groupId>org.apache.kudu</groupId>
            <artifactId>kudu-spark2_2.11</artifactId>
            <version>1.6.0</version>
        </dependency>

一、环境说明

集群:Spark 2.1.2 + hadoop2.3

开发机OS:win7

Jdk 1.8.0_151

下载jre即可 http://www.oracle.com/technetwork/java/javase/downloads/index.html

注:JDK是一个平台特定的软件,有针对Windows,Mac和Unix系统的不同的安装包。 可以说JDK是JRE的超集,它包含了JRE的Java编译器,调试器和核心类

scala 2.11.8 http://www.scala-lang.org/download/

IntelliJ IDEA 2017.3

spark 源码spark-2.1.2-bin-hadoop2.3

spark sql与mysql 和hdfs交互的实战

  • 1.添加jar包
  1. 正常配置不再赘述,这里如果需要读取MySQL数据,则需要在当前用户下的环境变量里额外加上JDBC的驱动jar包 例如我的是:mysql-connector-java-5.1.18-bin.jar 存放路径是$SPARK_HOME/jars 所以需要额外配置环境变量
    export PATH = $PATH:$SPARK_HOME/jars
  • 2.启动spark-shell

    bin/spark-shell --master=spark://h4:7077 --driver-class-path=./jars/mysql-connector-java-5.1.18-bin.jar -- jars=./jars/mysql-connector-java-5.1.18-bin.jar

  • 3.代码

spark-sql采用sql方式执行操作正常启动之后可以先通过spark-sql建立数据库并切换到当前新建的数据库
spark.sql("create database spark")
可以查看下是否新建成功
spark.sql("show databases ").show
创建成功之后切换数据库
spark.sql("use spark")
现在开始读取远程MySQL数据
val sql = """CREATE TABLE student USING org.apache.spark.sql.jdbc OPTIONS ( url "jdbc:mysql://worker2:3306/spark", dbtable "student", user "root", password "root" )"""
执行:
spark.sql(sql);

等待执行完毕之后,将表数据存入缓存
spark.sql("cache table student")
此时即可进行操作,例如:val studentDF = spark.sql("select id,name from student")
完成需求查询之后,可将结果以parquet的格式保存到HDFS
studentDF.write.parquet("hdfs://h4:9000/test/spark/parquet")
也可以写成json格式
studentDF.write.json("hdfs://h4:9000/test/spark/json")

  • 4.性能:

集群状态下,硬件配置32G内存 2T硬盘,spark配了4核,内存分配了20G的情况下,测试速度如下: 2700万条记录的表导入spark用时1秒以内 sparksql将其以json格式存入HDFS用时288秒,共1.0G,将其以parquet格式存入HDFS用时207秒,共86.6M,可见parquet的优势还是比较明显

参考链接: http://blog.51cto.com/10901776/1875371

本文用的是cloudera版本,添加:

二、环境搭建

 <repositories>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>

1. 基本配置

安装java、scala,配置环境变量JAVA_HOME、SCALA_HOME为对应安装路径

PATH后面添加%JAVA_HOME%jrebin; %SCALA_HOME%bin

【WIN】%JAVA_HOME%

【Linux】$JAVA_HOME

注意:scala 安装路径不能有空格,否则会报错

>>找不到或无法加载主类scala.tools.nsc.MainGenericRunner

检验标准

打开CMD,分别执行java、scala命令。

TAG标签:
版权声明:本文由金沙澳门官网4166发布于中国史,转载请注明出处:3开发环境搭建和开发示例,Spark的那些事