許多數(shù)據(jù)分析師都是用HIVE SQL跑數(shù)象迎，這里我建議轉(zhuǎn)向PySpark：
- PySpark的語法是從左到右串行的答恶，便于閱讀规阀、理解和修正魔熏；SQL的語法是從內(nèi)到外嵌套的衷咽，不方便維護(hù)；
- PySpark繼承Python優(yōu)美蒜绽、簡潔的語法镶骗，同樣的效果，代碼行數(shù)可能只有SQL的十分之一躲雅；
- Spark分轉(zhuǎn)化操作和行動操作鼎姊，只在行動操作時才真正計算，所以可以減少不必要的計算時間相赁；
- 相對于SQL層層嵌套的一個整體相寇，PySpark可以拆分成多步，并可以十分方便地把中間結(jié)果保存為變量钮科，更有利于調(diào)試和修改唤衫；
- PySpark可以與Python中的其他模塊結(jié)合使用，可以將多種功能有機結(jié)合成一個系統(tǒng)
- PySpark SQL模塊許多函數(shù)绵脯、方法與SQL中關(guān)鍵字一樣佳励，可以以比較低的學(xué)習(xí)成本切換
- 最重要的，Spark是基于內(nèi)存計算的桨嫁，計算速度本身比Hive快很多倍
PySpark的安裝配置
如果只是在單機上練習(xí)下植兰，照著網(wǎng)上的帖子在Linux系統(tǒng)安裝一下就可以了；如果想真正在集群上實戰(zhàn)璃吧，還是找運維搭建吧楣导。
PySpark SQL語法
最好的學(xué)習(xí)資料當(dāng)然是官方文檔，不過官方文檔是按函數(shù)名排序的畜挨，這對于新手不太友好筒繁，所以這里整理一下噩凹。
數(shù)據(jù)拉取
第一步是拉取數(shù)據(jù)，與SQL毡咏、Pandas驮宴、R一樣，在SparkSQL中呕缭，我們以DataFrame以基本的數(shù)據(jù)結(jié)構(gòu)（不過要注意堵泽，SparkSQL DataFrame與Pandas的DataFrame是兩種數(shù)據(jù)結(jié)構(gòu)，雖然相互轉(zhuǎn)換也很容易）恢总。
加載包
from __future__ import print_function
import pandas as pd
from pyspark.sql import HiveContext
from pyspark import SparkContext,SparkConf
from sqlalchemy import create_engine
import datetime
import pyspark.sql.functions as F
conf = SparkConf().setAppName("abc")
sc = SparkContext(conf=conf)
hiveCtx = HiveContext(sc)
手工創(chuàng)建一個DataFrame
d = [{'name': 'Alice', 'age': 1},{'name': 'Bob', 'age': 5}]
df = sqlContext.createDataFrame(d)
df.show()
從集群里運行SQL生成DataFrame
實際工作中往往是從集群中拉數(shù)迎罗，然后處理；還是執(zhí)行SQL（盡管仍是SQL片仿，但是不必寫復(fù)雜的SQL纹安；用基本的SQL先把源數(shù)據(jù)拉出來，復(fù)雜的處理和計算交給Spark來做）砂豌，以下是用Hive拉數(shù)：
sql = "" # 拉數(shù)SQL
df = hiveCtx.sql(sql)
緩存與清除緩存
Spark每次作行動操作時厢岂，都是從最初的轉(zhuǎn)化操作開始計算；如果不想從頭開始計算阳距，想保存中間結(jié)果表塔粒，就應(yīng)該把數(shù)據(jù)載入緩存。
df.cache()
與之相對的娄涩，清除緩存為
sqlContext.clearCache()
數(shù)據(jù)探索
展示
df.show() # 不加參數(shù)默認(rèn)展示前20行
統(tǒng)計行數(shù)
df.count()
查看schema
df.printSchema()
查看字段
df.columns
查看字段類型
df.dtypes
數(shù)據(jù)處理
查詢
df.select('age','name') # 帶show才能看到結(jié)果
別名
df.select(df.age.alias('age_value'),'name')
篩選
df.filter(df.name=='Alice')
增加列
增加列有2種方法窗怒，一種是基于現(xiàn)在的列計算映跟；一種是用pyspark.sql.functions的lit()增加常數(shù)列蓄拣。
df.select(df.age+1,'age','name')
df.select(F.lit(0).alias('id'),'age','name')
增加行
df.unionAll(df2)
刪除重復(fù)記錄
df.drop_duplicates()
去重
df.distinct()
刪除列
df.drop('id')
刪除存在缺失值的記錄
df.dropna(subset=['age', 'name']) # 傳入一個list，刪除指定字段中存在缺失的記錄
填補缺失值
df.fillna({'age':10,'name':'abc'}) # 傳一個dict進(jìn)去努隙，對指定的字段填充
分組計算
df.groupby('name').agg(F.max(df['age']))
join
df.groupby('name').agg(F.max(df['age']))
函數(shù)和UDF
pyspark.sql.functions里有許多常用的函數(shù)球恤，可以滿足日常絕大多數(shù)的數(shù)據(jù)處理需求；當(dāng)然也支持自己寫的UDF荸镊，直接拿來用咽斧。
自帶函數(shù)
根據(jù)官方文檔，以下是部分函數(shù)說明：
'lit': 'Creates a :class:`Column` of literal value.',
'col': 'Returns a :class:`Column` based on the given column name.',
'column': 'Returns a :class:`Column` based on the given column name.',
'asc': 'Returns a sort expression based on the ascending order of the given column name.',
'desc': 'Returns a sort expression based on the descending order of the given column name.',
'upper': 'Converts a string expression to upper case.',
'lower': 'Converts a string expression to upper case.',
'sqrt': 'Computes the square root of the specified float value.',
'abs': 'Computes the absolutle value.',
'max': 'Aggregate function: returns the maximum value of the expression in a group.',
'min': 'Aggregate function: returns the minimum value of the expression in a group.',
'first': 'Aggregate function: returns the first value in a group.',
'last': 'Aggregate function: returns the last value in a group.',
'count': 'Aggregate function: returns the number of items in a group.',
'sum': 'Aggregate function: returns the sum of all values in the expression.',
'avg': 'Aggregate function: returns the average of the values in a group.',
'mean': 'Aggregate function: returns the average of the values in a group.',
'sumDistinct': 'Aggregate function: returns the sum of distinct values in the expression.',
df.select(F.max(df.age))
df.select(F.min(df.age))
df.select(F.avg(df.age)) # 也可以用mean躬存，一樣的效果
df.select(F.countDistinct(df.age)) # 去重后統(tǒng)計
df.select(F.count(df.age)) # 直接統(tǒng)計张惹，經(jīng)試驗，這個函數(shù)會去掉缺失值會再統(tǒng)計
from pyspark.sql import Window
df.withColumn("row_number", F.row_number().over(Window.partitionBy("a","b","c","d").orderBy("time"))).show() # row_number()函數(shù)
日期相關(guān)函數(shù)參考：pyspark系列--日期函數(shù)
UDF
統(tǒng)計計算
描述性統(tǒng)計分析
df.describe('age').show() # describe()相當(dāng)于R里的summary()
數(shù)據(jù)寫出
數(shù)據(jù)寫出有以下幾種情況——
- 寫入集群分區(qū)表
all_bike.rdd.map(lambda line: u','.join(map(lambda x:unicode(x),line))).saveAsTextFile('/user/hive/warehouse/bi.db/bikeid_without_3codes_a_d/dt={}'.format(t0_uf)) #轉(zhuǎn)化為RDD寫入HDFS路徑
還有一種方法岭洲，是先把dataframe創(chuàng)建成一個臨時表宛逗，再用hive sql的語句寫入表的分區(qū)。
bike_change_2days.registerTempTable('bike_change_2days')
sqlContext.sql("insert into bi.bike_changes_2days_a_d partition(dt='%s') select citycode,biketype,detain_bike_flag,bike_tag_onday,bike_tag_yesterday,bike_num from bike_change_2days"%(date))
- 寫入集群非分區(qū)表
df_spark.write.mode("append").insertInto('bi.pesudo_bike_white_list') # 直接使用write.mode方法insert到指定的集群表
寫入數(shù)據(jù)庫
可以先將PySpark DataFrame轉(zhuǎn)化成Pandas DataFrame盾剩，然后用pandas的to_sql方法插入數(shù)據(jù)庫寫出本地
df.write.csv()
與Pandas DataFrame互相轉(zhuǎn)換
如果你熟悉Pandas包雷激，并且PySpark處理的中間數(shù)據(jù)量不是太大替蔬，那么可以直接轉(zhuǎn)換成pandas DataFrame，然后轉(zhuǎn)化成常規(guī)操作屎暇。
df.toPandas() # PySpark DataFrame轉(zhuǎn)化成Pandas DataFrame
import pandas as pd
df_p = pd.DataFrame(dict(num=range(3),char=['a','b','c']))
df_s = sqlContext.createDataFrame(df_p) # pandas dataframe轉(zhuǎn)化成PySpark DataFrame
type(df_s)
機器學(xué)習(xí)
關(guān)于機器學(xué)習(xí)承桥，在以后的文章里再單獨討論。
