Parquet介绍及简单使用（转）

// 读取 Parquet 文件
val usersDF = spark.read.load("/test/users.parquet")

// 查询 Schema 和数据
usersDF.printSchema
usersDF.show

// 查询用户的 name 和喜爱颜色并保存
usersDF.select($"name", $"favorite_color").write.save("/test/result/parquet")
// 验证结果 可通过 printSchema 查询数据结构，使用 show 查看数据

// 显式指定文件格式： 加载 json 格式
val usersDF = spark.read.format("json").load("/test/people.json")

// 存储模式(Save Modes) 
// 可以采用 SaveMode 执行存储操作， SaveMode 定义 了对数据的处理模式，需要注意的是，这些保存模式不使用任何锁定，不是原子操作
// 当使用 Overwrite 方式执行时，在输出新数据之前，原数据就已经被删除
usersDF.select($"name").write.save("/test/parquet1")   // 若 /test/parquet1 存在会报错
usersDF.select($"name").wirte.mode("overwrite").save("/test/parquet1")        // 使用 overwrite 即可

// 将结果保存为表, 也可以进行分区， 分桶等操作： partitionBy  bucketBy
usersDF.select($"name").write.saveAsTable("table1")

// 从 json 文件中读入数据
val empJson = spark.read.json("/test/emp.json")
// 将数据保存为 parquet
empJson.write.mode("overwrite").parquet("/test/parquet")
// 读取 parquet
val empParquet = spark.read.parquet("/test/parquet")
// 创建临时表 emptable
empParquet.createOrReplaceTempView("emptalbe")
// 使用 SQL 语句执行查询
spark.sql("select * from emptable where deptno=10 and sal>1500").show

// 创建第一个文件
val df1 = sc.makeRDD(1 to 5).map(x=> (x, x*2)).toDF("single", "double")
scala> df1.printSchema
root
 |-- single: integer (nullable = false)
 |-- double: integer (nullable = false)
 
 
// 创建第二个文件 
 scala> val df2 = sc.makeRDD(6 to 10).map(x=> (x, x*2)).toDF("single", "triple")
df2: org.apache.spark.sql.DataFrame = [single: int, triple: int]

scala> df2.printSchema
root
 |-- single: integer (nullable = false)
 |-- triple: integer (nullable = false)
  
 scala> df2.write.parquet("/data/testtable/key=2")

 // 合并上面的两个文件
scala> val df3 = spark.read.option("mergeSchema", "true").parquet("/data/testtable")
df3: org.apache.spark.sql.DataFrame = [single: int, double: int ... 2 more fields]

scala> df3.printSchema
root
 |-- single: integer (nullable = true)
 |-- double: integer (nullable = true)
 |-- triple: integer (nullable = true)
 |-- key: integer (nullable = true)
 
 scala> df3.show
+------+------+------+---+
|single|double|triple|key|
+------+------+------+---+
|     8|  null|    16|  2|
|     9|  null|    18|  2|
|    10|  null|    20|  2|
|     3|     6|  null|  1|
|     4|     8|  null|  1|
|     5|    10|  null|  1|
|     6|  null|    12|  2|
|     7|  null|    14|  2|
|     1|     2|  null|  1|
|     2|     4|  null|  1|
+------+------+------+---+

// 第一种
scala> val df4 = spark.read.json("/app/spark-2.2.1-bin-hadoop2.7/examples/src/main/resources/people.json")
df4: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> df4.show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

// 第二种
scala> val df5 = spark.read.format("json").load("/app/spark-2.2.1-bin-hadoop2.7/examples/src/main/resources/people.json")
df5: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

scala> df5.show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

// 将 JDBC 的驱动加入
bin/spark-shell --master spark://bigdata11:7077 --jars /root/temp/ojdbc6.jar --driver-class-path /root/temp/ojdbc6.jar

// 读取 Oracle
val oracleEmp = spark.read.format("jdbc")
                    .option("url","jdbc:oracle:thin:@192.168.10.100:1521/orcl.example.com")
                    .option("dbtable","scott.emp")
                    .option("user","scott")
                    .option("password","tiger").load

// 创建表
spark.sql("create table ccc(key INT, value STRING) row format delimited fields terminated by ‘,‘")

// 导入数据
spark.sql("load data local path ‘/test/data.txt‘ into table ccc")

// 查询数据
spark.sql("select * from ccc").show

show tables;
select * from ccc;