Spark SQL:结构化数据文件处理01

文章目录

准备工作掌握Dataframe基础操作

创建Dataframe对象

结构化数据文件创建Dataframe外部数据库创建DataframeRDD创建DataframeHive中的表创建Dataframe Dataframe查看数据

printSchema:打印数据模式show:查看数据frist/head/take/takeAsList:获取若干行记录collect/collectAsList:获取所有数据

准备工作

Linux安装Hive，MySQL
参考这位大佬的安装教程

掌握Dataframe基础操作 创建Dataframe对象 结构化数据文件创建Dataframe

hdfs dfs -mkdir /user/root/sparksql
hdfs dfs -put /home/xwk/software/spark/examples/src/main/resources/users.parquet /user/root/sparksql
hdfs dfs -put /home/xwk/software/spark/examples/src/main/resources/people.json /user/root/sparksql

scala> import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.SQLContext

scala> val sqlContext=new SQLContext(sc)
sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@438e1537

scala> val dfUsers=sqlContext.read.load("hdfs://master/user/root/sparksql/users.parquet")
dfUsers: org.apache.spark.sql.Dataframe = [name: string, favorite_color: string, favorite_numbers: array]

scala> val dfPeople=sqlContext.read.json("hdfs://master/user/root/sparksql/people.json")
dfPeople: org.apache.spark.sql.Dataframe = [age: bigint, name: string]

外部数据库创建Dataframe

前提是数据库和表都存在

scala> val url="jdbc:mysql://192.168.10.20:3306/hive"
url: String = jdbc:mysql://192.168.10.20:3306/hive

scala> val jdbcDF=sqlContext.read.format("jdbc").options(
     | Map("url"->url,
     | "user"->"root",
     | "password"->"123456",
     | "dbtable"->"DBS")).load()
jdbcDF: org.apache.spark.sql.Dataframe = [DB_ID: bigint, DESC: string, DB_LOCATION_URI: string, NAME: string, OWNER_NAME: string, OWNER_TYPE: string]

RDD创建Dataframe

scala> case class Person(name:String,age:Int)
defined class Person

scala> val data=sc.textFile("/user/root/sparksql/user.txt").map(_.split(","))
data: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[8] at map at :28

scala> val people=data.map(p=>Person(p(0),p(1).trim.toInt)).toDF()
people: org.apache.spark.sql.Dataframe = [name: string, age: int]

Hive中的表创建Dataframe

scala> import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.hive.HiveContext

scala> val hiveContext=new HiveContext(sc)
hiveContext: org.apache.spark.sql.hive.HiveContext = org.apache.spark.sql.hive.HiveContext@27f22d1a

scala> hiveContext.sql("use test")
res4: org.apache.spark.sql.Dataframe = [result: string]

scala> val people=hiveContext.sql("select * from students")
people: org.apache.spark.sql.Dataframe = [id: int, name: string, score: double, classes: string]

Dataframe查看数据

使用SparkContext读取该数据并转换为Dataframe

scala> case class Movie(movieId:Int,title:String,Genres:String)
defined class Movie

scala> val data=sc.textFile("hdfs://master/user/root/sparksql/movies.dat").map(_.split("::"))
data: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[20] at map at :34

scala> val movies=data.map(m=>Movie(m(0).trim.toInt,m(1),m(2))).toDF()
movies: org.apache.spark.sql.Dataframe = [movieId: int, title: string, Genres: string]

printSchema:打印数据模式

scala> movies.printSchema
root
 |-- movieId: integer (nullable = false)
 |-- title: string (nullable = true)
 |-- Genres: string (nullable = true)

show:查看数据

下面这个控制的是字符不是行数，可以看到所有字符都显示出来了

frist/head/take/takeAsList:获取若干行记录

scala> movies.first //获取第一行记录
res14: org.apache.spark.sql.Row = [1,Toy Story (1995),Animation|Children's|Comedy]

scala> movies.head(2) //获取前2行记录
res15: Array[org.apache.spark.sql.Row] = Array([1,Toy Story (1995),Animation|Children's|Comedy], [2,Jumanji (1995),Adventure|Children's|Fantasy])

scala> movies.take(2) //获取前2行记录
res16: Array[org.apache.spark.sql.Row] = Array([1,Toy Story (1995),Animation|Children's|Comedy], [2,Jumanji (1995),Adventure|Children's|Fantasy])

scala> movies.takeAsList(2) //获取前2行记录，并以List的形式展现
res17: java.util.List[org.apache.spark.sql.Row] = [[1,Toy Story (1995),Animation|Children's|Comedy], [2,Jumanji (1995),Adventure|Children's|Fantasy]]

collect/collectAsList:获取所有数据

movies.collect()
movies.collectAsList()