飞谷云 MySQL 第3讲
MySQL+表连接（下）
一、MySQL+表连接（下）
1、内连接（inner join）：结果仅满足条件的行
select * from T1,T2 where T1.stud_code=T2.stud_code;
select * from T1 inner join T2 ON T1.stud_code=T2.stud_code;#效率比上句高
2、左连接（left outer join）:
显示左表T1中的所有行，并把右表T2中符合条件加到左表T1中；
右表T2中不符合条件，就不用加入结果表中，并且NULL表示。
select * from T1 left outer join T2 ON T1.stud_code=T2.stud_code;
3、右连接（right outer join）
显示右表T2中的所有行，并把左表T1中符合条件加到右表T2中；
左表T1中不符合条件，就不用加入结果表中，并且NULL表示。
4、全连接（full outer join）
显示左表T1、右表T2两边中的所有行，即把左联结果表 + 右联结果表组合在一起，然后过滤掉重复的。
/*内连接*/
select * from stud_info T1 ,stud_score T2 WHERE T1.stud_code=T2.stud_code;
select * from stud_info T1 inner join stud_score T2 ON T1.stud_code=T2.stud_code;
/*左连接*/
select * from stud_info T1 LEFT OUTER join stud_score T2 ON T1.stud_code=T2.stud_code;
/*右连接*/
select * from stud_info T1 RIGHT OUTER join stud_score T2 ON T1.stud_code=T2.stud_code;
二、pandas 中的 DataFrame的表连接
- 导入必要的包或模块
import numpy as np
import pandas as pd
from pandas import DataFrame
import MySQLdb
两个DataFrame数据集如何进行关系？
df1=DataFrame({'key':['a','a','b','c','c'],'data1':range(5)})
df2=DataFrame({'key':['a','b','d'],'data1':range(3)})
- 内连接
df3=pd.merge(df1,df2,on='key')
df4=pd.merge(df1,df2,on='key',how='inner')
df5=pd.merge(df1,df2,left_on='key',right_on='key',how='inner')
#左连接
df_l=pd.merge(df1,df2,on='key',how='left')
#右连接
df_r=pd.merge(df1,df2,on='key',how='right')
#全连接
df_a=pd.merge(df1,df2,on='key',how='outer')
- 两个表结构相同字段，自动添加后缀，_x,_y,我们也可以自定义，后缀，通过
suffixes=(‘_left’，‘_right’)
df6=pd.merge(df1,df2,on='key',how='inner',suffixes=('_left','_right'))
#选择数据
df7=df6[['key','data1_right']]
#修改列名（data1_right改为data）
df8=df7.rename(columns={'data1_right':'data'})
#保存结果
df8.to_csv('data07.csv')
df8.to_csv('data07.csv',index=False,cols=['key','data'])##忽略行索引
- 把MySQL数据库表导入pandas
conn= MySQLdb.connect(host='localhost',port=3306,user='feigu_mysql', passwd='feigu2016', db='testdb',charset='utf8')
data2 = pd.read_sql('select * from stud_info', conn)
- 如果文件较大，也可可以逐块读，如：chunksize=1000（行）
`inputfile1 = '/home/feigu/python_test/data/stud_score.csv'`
#输入的数据文件
`inputfile2 = '/home/feigu/python_test/data/stud_info.csv'`
#输入的数据文件
data1 = pd.read_csv(inputfile1) #读取数据
data2 = pd.read_csv(inputfile2) #读取数据
df1=DataFrame(data1) #转换为DataFrame格式
df2=DataFrame(data2) #转换为DataFrame格式
#如果要跳过一些行，可以加上参数，skiprow[0,2]
pd.read_csv(inputfile2,skiprow[0,2])跳过第1行，第3行
df.count() #查看非NA值得数量
df.columns #查看列名
df.<tab> #查看df可以使用的方法
df.head(2) #查看前2行
df.tail(2) #查看倒数2行
三、把数据导入Spark-SQL中的DataFrame结构中
#把数据导入spark-rdd
#进入交互式编程环境
spark-shell --master local[2]
==========使用scala语言==========
// 首先用已有的Spark Context对象创建SQLContext对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// 导入语句，可以隐式地将RDD转化成DataFrame
import sqlContext.implicits._
// 创建一个表示stud的自定义类
case class stud_score(stud_code:String, sub_code:String, sub_name: String, sub_tech: String, sub_score: Int)
//用文件创建一个RDD
val rdd_st = sc.textFile("file:///home/hadoop/data/stud_score.csv")
//转换为含case的rdd
val rdd_case=rdd_st.map(_.split(",")).map(p => stud_score(p(0), p(1), p(2), p(3), p(4).trim.toInt))
//把RDD转换为DataFrame
val dfstud=rdd_case.toDF()
// 将DataFrame注册为一个表
dfstud.registerTempTable("stud_score")
// 用sqlContext对象提供的sql方法执行SQL语句。
val result80 = sqlContext.sql("SELECT * FROM stud_score where sub_score>=80")
//显示结果
result80.show()
result80.take(6)
//也可以保存
result80.rdd.saveAsTextFile("/home/hadoop/data")
