介紹

DataFrame是Spark推薦的統(tǒng)一結(jié)構(gòu)化數(shù)據(jù)接口，基于DataFrame快速實(shí)現(xiàn)結(jié)構(gòu)化數(shù)據(jù)的分析，詳細(xì)使用教程在https://spark.apache.org/docs/latest/sql-programming-guide.html

使用

創(chuàng)建SparkQL的上下文。

from pyspark.sqlimportSQLContext

sqlContext = SQLContext(sc)

導(dǎo)入JSON文件數(shù)據(jù)，DataFrame也支持從RDD、JDBC、Hive等數(shù)據(jù)源導(dǎo)入數(shù)據(jù)。

df = sqlContext.read.json("/tmp/git.json")

git.json的數(shù)據(jù)格式類似這樣，可以通過(guò)git log --pretty=format:'{"commit":"%H","author":"%an","author_email":"%ae","date":"%ad"}' > git.json來(lái)生成。

{"commit":"fbbf4b22db7857f11018f0153472d909af874c31","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Fri Jan 1 09:47:31 2016 +0800"}

{"commit":"22ef72a98c9dfe2f63db9cf34c635124b2d61676","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Wed Dec 30 15:04:16 2015 +0800"}

{"commit":"1c6f4826526149d1df4d6f49c4cd54def5c09ec0","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Wed Dec 30 14:59:18 2015 +0800"}

{"commit":"56b4161ff9913033fe0dcdf291eca9ec0a6a9cc5","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Wed Dec 30 09:19:56 2015 +0800"}

{"commit":"0c8c9b065ad381362cbe6726df09b939796175ae","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Tue Dec 29 15:10:43 2015 +0800"}

{"commit":"b4e784bf78a83a922cff31de239c21b168bc7809","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Tue Dec 29 15:09:58 2015 +0800"}

{"commit":"2e02e17465c2594defb81c439bffe3a3a63ddf92","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Mon Dec 28 20:12:24 2015 +0800"}

{"commit":"185507c50f91a32172a106dd2d1b2fba5cab129c","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Sun Nov 29 22:47:18 2015 +0800"}

{"commit":"512761a255619d6dc81c4ba2d892d397b390b978","author":"tobe","author_email":"tobeg3oogle@gmail.com","date":"Sun Nov 29 21:59:29 2015 +0800"}

基本操作。

df.show()

df.printSchema()

df.select("author").show()

df.filter(df['author'] !="tobe").show()

df.groupBy("author").count().show()

執(zhí)行SQL命令。

df.registerTempTable("git")

df = sqlContext.sql("SELECT * FROM git").show()

通過(guò)代碼創(chuàng)建DataFrame。

anotherPeopleRDD = sc.parallelize(['{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}'])

anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)

準(zhǔn)備MySQL數(shù)據(jù)庫(kù)。

apt-get install -y libmysql-java

mysql -uroot -p

create database spark_db;

use spark_db;

create table spark_table (name varchar(20), ageint(32));

insert into spark_table values ("tobe",18);

insert into spark_table values ("john",28);

連接MySQL。

SPARK_CLASSPATH=/usr/share/java/mysql-connector-java.jar ./pyspark

from pyspark.sqlimportSQLContext

sqlContext = SQLContext(sc)

dataframe_mysql = sqlContext.read.format("jdbc").options(url="jdbc:mysql://127.0.0.1:3306/spark_db", driver="com.mysql.jdbc.Driver", dbtable="spark_table", user="root", password="root").load()

dataframe_mysql.show()