pyspark訪問hive數據實戰
數據分析都是直接使用hive腳本進行調用，隨著APP用戶行為和日志數據量的逐漸累積，跑每天的腳本運行需要花的時間越來越長，雖然進行了sql優化，但是上spark已經提上日程。
直接進行spark開發需要去學習scala，為了降低數據分析師的學習成本，決定前期先試用sparkSQL，能夠讓計算引擎無縫從MR切換到spark，現在主要使用pyspark訪問hive數據。
以下是安裝配置過程中的詳細步驟：
1.安裝spark
需要先安裝JDK和scala，這不必多說，由于現有hadoop集群版本是采用的2.6.3，所以spark版本是下載的穩定版本spark-1.4.0-bin-hadoop2.6.tgz
我是先在一臺機器上完成了Spark的部署，Master和Slave都在一臺機器上。注意要配置免秘鑰ssh登陸。
1.1 環境變量配置
- export JAVA_HOME=/usr/jdk1.8.0_73
- export HADOOP_HOME=/usr/hadoop
- export HADOOP_CONF_DIR=/usr/hadoop/etc/hadoop
- export SCALA_HOME=/usr/local/scala-2.11.7
- export SPARK_HOME=/home/hadoop/spark_folder/spark-1.4.0-bin-hadoop2.6
- export SPARK_MASTER_IP=127.0.0.1
- export SPARK_MASTER_PORT=7077
- export SPARK_MASTER_WEBUI_PORT=8099
- export SPARK_WORKER_CORES=3 //每個Worker使用的CPU核數
- export SPARK_WORKER_INSTANCES=1 //每個Slave中啟動幾個Worker實例
- export SPARK_WORKER_MEMORY=10G //每個Worker使用多大的內存
- export SPARK_WORKER_WEBUI_PORT=8081 //Worker的WebUI端口號
- export SPARK_EXECUTOR_CORES=1 //每個Executor使用使用的核數
- export SPARK_EXECUTOR_MEMORY=1G //每個Executor使用的內存
- export HIVE_HOME=/home/hadoop/hive
- export SPARK_CLASSPATH=$HIVE_HOME/lib/mysql-connector-java-5.1.31-bin.jar:$SPARK_CLASSPATH
- export LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:$HADOOP_HOME/lib/native
1.2 配置slaves
- cp slaves.template slaves
- vi slaves 添加以下內容：localhost
1.3 啟動master和slave
- cd $SPARK_HOME/sbin/
- ./start-master.sh
- 啟動日志位于 $SPARK_HOME/logs/目錄，訪問 http://localhost:8099，即可看到Spark的WebUI界面
- 執行 ./bin/spark-shell,打開Scala到Spark的連接窗口
2.SparkSQL與Hive的整合
- 拷貝$HIVE_HOME/conf/hive-site.xml和hive-log4j.properties到 $SPARK_HOME/conf/
- 在$SPARK_HOME/conf/目錄中，修改spark-env.sh，添加
- export HIVE_HOME=/home/hadoop/hive
- export SPARK_CLASSPATH=$HIVE_HOME/lib/mysql-connector-java-5.1.31-bin.jar:$SPARK_CLASSPATH
- 另外也可以設置一下Spark的log4j配置文件，使得屏幕中不打印額外的INFO信息(如果不想受干擾可設置為更高):
- log4j.rootCategory=WARN, console
- 進入$SPARK_HOME/bin，執行 ./spark-sql –master spark://127.0.0.1:7077 進入spark-sql CLI:
- [hadoop@hadoop spark]$ bin/spark-sql --help
- Usage: ./bin/spark-sql [options] [cli option]
- CLI options:
- -d,--define <keykey=value> Variable subsitution to apply to hive
- commands. e.g. -d A=B or --define A=B
- --database <databasename> Specify the database to use
- -e <quoted-query-string> SQL from command line
- -f <filename> SQL from files
- -h <hostname> connecting to Hive Server on remote host
- --hiveconf <propertyproperty=value> Use value for given property
- --hivevar <keykey=value> Variable subsitution to apply to hive
- commands. e.g. --hivevar A=B
- -i <filename> Initialization SQL file
- -p <port> connecting to Hive Server on port number
- -S,--silent Silent mode in interactive shell
- -v,--verbose Verbose mode (echo executed SQL to the
- console)
需要注意的是CLI不是使用JDBC連接，所以不能連接到ThriftServer;但可以配置conf/hive-site.xml連接到hive的metastore，然后對hive數據進行查詢。下面我們接著說如何在python中連接hive數據表查詢。
3.配置pyspark和示例代碼
3.1 配置pyspark
- 打開/etc/profile:
- #PythonPath 將Spark中的pySpark模塊增加的Python環境中
- export PYTHONPATH=/opt/spark-hadoop/python
- source /etc/profile
執行./bin/pyspark ,打開Python到Spark的連接窗口，確認沒有報錯。
打開命令行窗口，輸入python,Python版本為2.7.6，如圖所示，注意Spark暫時不支持Python3。輸入import pyspark不報錯，證明開發前工作已經完成。
3.2 啟動ThriftServer
啟動ThriftServer，使之運行在spark集群中：
sbin/start-thriftserver.sh --master spark://localhost:7077 --executor-memory 5g
ThriftServer可以連接多個JDBC/ODBC客戶端，并相互之間可以共享數據。
3.3 請求示例
查看spark官方文檔說明，spark1.4和2.0對于sparksql調用hive數據的API變化并不大。都是用sparkContext 。
- from pyspark import SparkConf, SparkContext
- from pyspark.sql import HiveContext
- conf = (SparkConf()
- .setMaster("spark://127.0.0.1:7077")
- .setAppName("My app")
- .set("spark.executor.memory", "1g"))
- sc = SparkContext(conf = conf)
- sqlContext = HiveContext(sc)
- my_dataframe = sqlContext.sql("Select count(1) from logs.fmnews_dim_where")
- my_dataframe.show()
返回結果：
運行以后在webUI界面看到job運行詳情。
4.性能比較
截取了接近一個月的用戶行為數據，數據大小為2G，總共接近1600w條記錄。
為了測試不同sql需求情況下的結果，我們選取了日常運行的2類sql：
1.統計數據條數：
- select count(1) from fmnews_user_log2;
2.統計用戶行為：
- SELECT device_id, min_time FROM
- (SELECT device_id,min(import_time) min_time FROM fmnews_user_log2
- GROUP BY device_id)a
- WHERE from_unixtime(int(substr(min_time,0,10)),'yyyy-MM-dd') = '2017-03-02';
3. 用戶行為分析：
- select case when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '06:00' and '07:59' then 1
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '08:00' and '09:59' then 2
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '10:00' and '11:59' then 3
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '12:00' and '13:59' then 4
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '14:00' and '15:59' then 5
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '16:00' and '17:59' then 6
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '18:00' and '19:59' then 7
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '20:00' and '21:59' then 8
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '22:00' and '23:59' then 9
- else 0 end fmnews_time_type, count(distinct device_id) device_count,count(1) click_count
- from fmcm.fmnews_user_log2
- where from_unixtime(int(substr(import_time,0,10)),'yyyy-MM-dd') = '2017-03-02'
- group by case when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '06:00' and '07:59' then 1
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '08:00' and '09:59' then 2
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '10:00' and '11:59' then 3
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '12:00' and '13:59' then 4
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '14:00' and '15:59' then 5
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '16:00' and '17:59' then 6
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '18:00' and '19:59' then 7
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '20:00' and '21:59' then 8
- when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '22:00' and '23:59' then 9
- else 0 end;
第一條sql的執行結果對比：hive 35.013 seconds
第一條sql的執行結果對比：sparksql 1.218 seconds
第二條sql的執行結果對比：hive 78.101 seconds
第二條sql的執行結果對比：sparksql 8.669 seconds
第三條sql的執行結果對比：hive 101.228 seconds
第三條sql的執行結果對比：sparksql 14.221 seconds
可以看到，雖然沒有官網吹破天的100倍性能提升，但是根據sql的復雜度來看10~30倍的效率還是可以達到的。
不過這里要注意到2個影響因子：
1. 我們數據集并沒有采取全量，在數據量達到TB級別兩者的差距應該會有所減小。同時sql也沒有針對hive做優化。
2. spark暫時是單機(內存足夠)并沒有搭建集群，hive使用的hadoop集群有4臺datanode。
