spark 2.X与1.x的区别
spark sql 2.x以上版本和1.x版本有个很大的区别:spark1.x的sqlContext在spark2.0中被整合到sparkSession,故而利用spark-shell客户端操作会有些许不同,具体如下文所述。
载入外部数据的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")
spark sql与mysql 和hdfs交互的实战
- 1.添加jar包
- 正常配置不再赘述,这里如果需要读取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的优势还是比较明显