SparkDataFrame入门教程

介绍

DataFrame是Spark推荐的统一结构化数据接口，基于DataFrame快速实现结构化数据的分析，详细使用教程在https://spark.apache.org/docs/latest/sql-programming-guide.html

使用

创建SparkQL的上下文。

from pyspark.sqlimportSQLContext

sqlCOntext= SQLContext(sc)

导入JSON文件数据，DataFrame也支持从RDD、JDBC、Hive等数据源导入数据。

df = sqlContext.read.json(&＃8220;/tmp/git.json&＃8221;)

git.json的数据格式类似这样，可以通过git log &＃8211;pretty=format:'{&＃8220;commit&＃8221;:&＃8221;%H&＃8221;,&＃8221;author&＃8221;:&＃8221;%an&＃8221;,&＃8221;author_email&＃8221;:&＃8221;%ae&＃8221;,&＃8221;date&＃8221;:&＃8221;%ad&＃8221;}&＃8217; > git.json来生成。

{&＃8220;commit&＃8221;:&＃8221;fbbf4b22db7857f11018f0153472d909af874c31&＃8243;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Fri Jan 1 09:47:31 2016 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;22ef72a98c9dfe2f63db9cf34c635124b2d61676&＃8243;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Wed Dec 30 15:04:16 2015 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;1c6f4826526149d1df4d6f49c4cd54def5c09ec0&＃8243;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Wed Dec 30 14:59:18 2015 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;56b4161ff9913033fe0dcdf291eca9ec0a6a9cc5&＃8243;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Wed Dec 30 09:19:56 2015 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;0c8c9b065ad381362cbe6726df09b939796175ae&＃8221;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Tue Dec 29 15:10:43 2015 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;b4e784bf78a83a922cff31de239c21b168bc7809&＃8243;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Tue Dec 29 15:09:58 2015 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;2e02e17465c2594defb81c439bffe3a3a63ddf92&＃8243;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Mon Dec 28 20:12:24 2015 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;185507c50f91a32172a106dd2d1b2fba5cab129c&＃8221;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Sun Nov 29 22:47:18 2015 +0800&＃8243;}

{&＃8220;commit&＃8221;:&＃8221;512761a255619d6dc81c4ba2d892d397b390b978&＃8243;,&＃8221;author&＃8221;:&＃8221;tobe&＃8221;,&＃8221;author_email&＃8221;:&＃8221;tobeg3oogle@gmail.com&＃8221;,&＃8221;date&＃8221;:&＃8221;Sun Nov 29 21:59:29 2015 +0800&＃8243;}

基本操作。

df.show()

df.printSchema()

df.select(&＃8220;author&＃8221;).show()

df.filter(df[&＃8216;author&＃8217;] !=&＃8221;tobe&＃8221;).show()

df.groupBy(&＃8220;author&＃8221;).count().show()

执行SQL命令。

df.registerTempTable(&＃8220;git&＃8221;)

df = sqlContext.sql(&＃8220;SELECT * FROM git&＃8221;).show()

通过代码创建DataFrame。

anotherPeopleRDD = sc.parallelize([&＃8216;{&＃8220;name&＃8221;:&＃8221;Yin&＃8221;,&＃8221;address&＃8221;:{&＃8220;city&＃8221;:&＃8221;Columbus&＃8221;,&＃8221;state&＃8221;:&＃8221;Ohio&＃8221;}}&＃8217;])

anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)

准备MySQL数据库。

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 (&＃8220;tobe&＃8221;,18);

insert into spark_table values (&＃8220;john&＃8221;,28);

连接MySQL。

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

from pyspark.sqlimportSQLContext

sqlCOntext= SQLContext(sc)

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

dataframe_mysql.show()