SparkSQL操作外部数据源

• parquet数据
• hive表数据
• mysql表数据
• hive与mysql结合

1.处理parquet数据

• 启动spark-shell:
  spark-shell --master local[2] --jars ~/software/mysql-connector-java-5.1.27-bin.jar

在spark-shell模式下，执行 标准的加载方法 :

val path = "file:///home/hadoop/app/xxx.parquet"//处理的parquet文件的路径
val userDF = spark.read.format("parquet").load(path)
userDF.printSchema()//打印DataFrame的Schema
userDF.show()//显示数据
userDF.select("name","favorite_color").show//选择性的显示两列
userDF.select("name","favorite_color").write.format("json").save("file:///home/hadoop/tmp/jsonout")//将查询到的数据以json形式写入到指定路径下

第二种加载parquet文件的方法，不指定文件format：
spark.read.load("file:///home/hadoop/app/users.parquet").show
第三种加载文件方法，option：
spark.read.format("parquet").option("path", "file:///home/hadoop/app/users.parquet")

• 注意，load方法默认加载的文件形式是parquet
  image.png

比如，下面这样，使用load方法处理一个parquet文件，不指定文件形式：

val userDF = spark.read.load("file:///home/hadoop/app/spark/examples/src/main/resources/people.json")

报错信息：

RuntimeException: file:/home/hadoop/app/spark-2.1.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/people.json is not a Parquet file

• 也可以进入sql模式下通过表来操作文件，执行spark-sql:
  spark-sql --master local[2] --jars ~/software/mysql-connector-java-5.1.27-bin.jar

CREATE TEMPORARY VIEW parquetTable
USING org.apache.spark.sql.parquet
OPTIONS (
  path "/home/hadoop/app/spark-2.2.0-bin-2.6.0-cdh5.7.0/examples/src/main/resources/users.parquet"
)

SELECT * FROM parquetTable

2.操作hive表数据

在spark-shell模式下，

spark.sql("show tables").show //显示表
spark.table("emp").show  //显示emp表的数据
spark.sql("select empno,count(1) from emp group by empno").show //按照empno分组显示
spark.sql("select empno,count(1) from emp group by empno").filter("empno is not null").write.saveAsTable("emp_1") //按照empno分组且过滤掉null的行，然后存储到hive表里

然而，执行下面的语句时，

spark.sql("select empno,count(1) from emp group by empno").filter("empno is not null").write.saveAsTable("emp_1")

报错：

org.apache.spark.sql.AnalysisException: Attribute name "count(1)" contains invalid character(s) among " ,;{}()\n\t=". Please use alias to rename it.;

需要加上别名才能存储到hive表中

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

在生产环境中要注意设置spark.sql.shuffle.partitions，默认是200

spark.sqlContext.setConf("spark.sql.shuffle.partitions","10")
spark.sqlContext.getConf("spark.sql.shuffle.partitions") //结果为10

image.png

当然也可以访问SparkUI页面的jobs标签页，查看相关信息。

3.操作mysql数据（替代Sqoop）

scala实现：

spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/hive").option("dbtable", "hive.TBLS").option("user", "root").option("password", "root").option("driver", "com.mysql.jdbc.Driver").load()

• 注意：如果没有driver option的话，会报错：java.sql.SQLException: No suitable driver

java实现:

import java.util.Properties
val connectionProperties = new Properties()
connectionProperties.put("user", "root")
connectionProperties.put("password", "root")
connectionProperties.put("driver", "com.mysql.jdbc.Driver")

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

spark-sql实现:

CREATE TEMPORARY VIEW jdbcTable
USING org.apache.spark.sql.jdbc
OPTIONS (
  url "jdbc:mysql://localhost:3306",
  dbtable "hive.TBLS",
  user 'root',
  password 'root',
  driver 'com.mysql.jdbc.Driver'
)

4.hive和mysql数据源数据查询

由于hive加载的数据，和mysql加载的数据源，都可以抽象为DataFrame，所以，不同的数据源可以通过DataFrame的select，join方法来处理显示。

5. third-party packages

A community index of third-party packages for Apache Spark
https://spark-packages.org/?q=tags%3A%22Data%20Sources%22