目的是將phoenix做存儲,spark做計算層。這樣就結合了phoenix查詢速度快和spark計算速度快的優點。
在這里將Phoenix的表作為spark的RDD或者DataFrames來操作,並且將操作的結果寫回phoenix中。
這樣做也擴大了兩者的使用場景。
Phoenix 版本 4.4.0
Hbase版本 0.98
spark版本 spark-1.5.2-bin-hadoop2.6
首先配置 SPARK_CLASSPATH
要想在spark中操作phoenix,就必須讓spark可以找到phoenix的相關類,所以我們把client放到spark_classpath中
export SPARK_CLASSPATH=$SPARK_CLASSPATH:/home/hadoop/phoenix/phoenix-spark-4.4.0-HBase-0.98-tests.jar export SPARK_CLASSPATH=$SPARK_CLASSPATH:/home/hadoop/phoenix/phoenix-4.4.0-HBase-0.98-client.jar export SPARK_CLASSPATH=$SPARK_CLASSPATH:/home/hadoop/phoenix/phoenix-server-client-4.4.0-HBase-0.98.jar
這樣就可以在spark-shell中操作phoenix了
下來結合兩者做下實驗:
1> 在phoenix中創建幾張表
[hadoop@10.10.113.45 ~/phoenix/bin]$>./sqlline.py 10.10.113.45:2181 0: jdbc:phoenix:10.10.113.45:2181> CREATE TABLE EMAIL_ENRON( . . . . . . . . . . . . . . . . .> MAIL_FROM BIGINT NOT NULL, . . . . . . . . . . . . . . . . .> MAIL_TO BIGINT NOT NULL . . . . . . . . . . . . . . . . .> CONSTRAINT pk PRIMARY KEY(MAIL_FROM, MAIL_TO)); 0: jdbc:phoenix:10.10.113.45:2181> CREATE TABLE EMAIL_ENRON_PAGERANK( . . . . . . . . . . . . . . . . .> ID BIGINT NOT NULL, . . . . . . . . . . . . . . . . .> RANK DOUBLE . . . . . . . . . . . . . . . . .> CONSTRAINT pk PRIMARY KEY(ID)); No rows affected (0.52 seconds)
查看下是否創建成功
0: jdbc:phoenix:10.10.113.45:2181> !tables +------------------------------------------+------------------------------------------+------------------------------------------+--------------+ | TABLE_CAT | TABLE_SCHEM | TABLE_NAME | | +------------------------------------------+------------------------------------------+------------------------------------------+--------------+ | | SYSTEM | CATALOG | SYSTEM TABLE | | | SYSTEM | FUNCTION | SYSTEM TABLE | | | SYSTEM | SEQUENCE | SYSTEM TABLE | | | SYSTEM | STATS | SYSTEM TABLE | | | | EMAIL_ENRON | TABLE | | | | EMAIL_ENRON_PAGERANK | TABLE | +------------------------------------------+------------------------------------------+------------------------------------------+--------------+ 0: jdbc:phoenix:10.10.113.45:2181>
2> 在將數據load到phoenix中,數據有40萬行
[hadoop@10.10.113.45 ~/phoenix/bin]$>./psql.py -t EMAIL_ENRON 10.10.113.45:2181 /home/hadoop/sfs/enron.csv SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. 15/12/03 10:06:37 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable csv columns from database. CSV Upsert complete. 367662 rows upserted Time: 21.783 sec(s)
數據來源:https://snap.stanford.edu/data/email-Enron.html
然后在查詢下
0: jdbc:phoenix:10.10.113.45:2181> select count(*) from EMAIL_ENRON; +------------------------------------------+ | COUNT(1) | +------------------------------------------+ | 367662 | +------------------------------------------+ 1 row selected (0.289 seconds)
看37萬數據,查詢不到一秒!!!
下面進入到spark-shell 的交互模式,我們做一個PageRank 算法的例子
[hadoop@10.10.113.45 ~/spark/bin]$>./spark-shell scala> import org.apache.spark.graphx._ import org.apache.spark.graphx._ scala> import org.apache.phoenix.spark._ import org.apache.phoenix.spark._ scala> val rdd = sc.phoenixTableAsRDD("EMAIL_ENRON", Seq("MAIL_FROM", "MAIL_TO"), zkUrl=Some("10.10.113.45")) rdd: org.apache.spark.rdd.RDD[Map[String,AnyRef]] = MapPartitionsRDD[2] at map at SparkContextFunctions.scala:39 scala> val rawEdges = rdd.map{ e => (e("MAIL_FROM").asInstanceOf[VertexId], e("MAIL_TO").asInstanceOf[VertexId]) } rawEdges: org.apache.spark.rdd.RDD[(org.apache.spark.graphx.VertexId, org.apache.spark.graphx.VertexId)] = MapPartitionsRDD[3] at map at <console>:29 scala> val graph = Graph.fromEdgeTuples(rawEdges, 1.0) graph: org.apache.spark.graphx.Graph[Double,Int] = org.apache.spark.graphx.impl.GraphImpl@621bb3c3 scala> val pr = graph.pageRank(0.001) pr: org.apache.spark.graphx.Graph[Double,Double] = org.apache.spark.graphx.impl.GraphImpl@55e444b1 scala> pr.vertices.saveToPhoenix("EMAIL_ENRON_PAGERANK", Seq("ID", "RANK"), zkUrl = Some("10.10.113.45"))(這一步會很耗內存,可能有的同學在測試的時候會報OOM,建議增大spark中executor memory,driver memory的大小)
我們在去phoenix中查看一下結果。
0: jdbc:phoenix:10.10.113.45:2181> select count(*) from EMAIL_ENRON_PAGERANK; +------------------------------------------+ | COUNT(1) | +------------------------------------------+ | 29000 | +------------------------------------------+ 1 row selected (0.113 seconds) 0: jdbc:phoenix:10.10.113.45:2181> SELECT * FROM EMAIL_ENRON_PAGERANK ORDER BY RANK DESC LIMIT 5; +------------------------------------------+------------------------------------------+ | ID | RANK | +------------------------------------------+------------------------------------------+ | 273 | 117.18141799210386 | | 140 | 108.63091596789913 | | 458 | 107.2728800448782 | | 588 | 106.11840798585399 | | 566 | 105.13932886531066 | +------------------------------------------+------------------------------------------+ 5 rows selected (0.568 seconds)